diff --git a/deploy/sql/海外区域统一增量-MySQL8.sql b/deploy/sql/海外区域统一增量-MySQL8.sql index 11ae9469b..96ff3dd7a 100644 --- a/deploy/sql/海外区域统一增量-MySQL8.sql +++ b/deploy/sql/海外区域统一增量-MySQL8.sql @@ -1990,6 +1990,19 @@ DELIMITER ; CALL assert_tail_order_commands_ready(); DROP PROCEDURE IF EXISTS assert_tail_order_commands_ready; +CREATE TABLE IF NOT EXISTS ll_container_plan_conversion_command ( + id BIGINT NOT NULL COMMENT '柜级转换命令ID', source_container_booking_id VARCHAR(128) NOT NULL COMMENT '稳定来源柜预订ID', + idempotency_key VARCHAR(128) NOT NULL COMMENT '首次调用幂等键', request_hash CHAR(64) NOT NULL COMMENT '确认请求SHA-256摘要', + snapshot_version BIGINT NOT NULL COMMENT '确认的柜级快照版本', snapshot_hash CHAR(64) NOT NULL COMMENT '确认的柜级快照SHA-256', + warehouse_id BIGINT NOT NULL COMMENT '海外接柜仓ID', batch_key VARCHAR(160) NOT NULL COMMENT '海外仓托盘批次幂等键', + status VARCHAR(32) NOT NULL COMMENT 'PREPARED/APPLYING/LOCAL_COMMITTED/COMPLETED', preparation_json JSON NOT NULL COMMENT '稳定计划号与托盘业务键快照', + result_json JSON NULL COMMENT '本地计划提交结果', last_error VARCHAR(1000) NULL COMMENT '最近一次跨服务错误', + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY(id), UNIQUE KEY uk_ll_container_conversion_command(source_container_booking_id), + UNIQUE KEY uk_ll_container_conversion_batch(warehouse_id,batch_key), KEY idx_ll_container_conversion_status(status,updated_at) +) ENGINE=InnoDB COMMENT='柜级入库计划转换Saga与幂等命令'; +ALTER TABLE ll_inbound_plan_pallet MODIFY COLUMN pallet_no VARCHAR(128) NOT NULL COMMENT 'Warehouse权威实物托盘码'; + SELECT 'PASS' AS migration_status, '海外尾程人工计划、暂扣与人工节点结构已就绪' AS message; -- 柜级入库计划跨客户批量托盘预留、确认与失败补偿。 diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanController.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanController.java index a18359725..7930f219a 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanController.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanController.java @@ -16,6 +16,7 @@ import com.baiyue.lastleg.inboundplan.dto.UpdateFbaAppointmentRequest; import com.baiyue.lastleg.inboundplan.dto.UpdateInboundPlanRequest; import com.baiyue.lastleg.inboundplan.service.InboundFbaAppointmentService; import com.baiyue.lastleg.inboundplan.service.InboundPlanService; +import com.baiyue.lastleg.inboundplan.service.ContainerInboundPlanConversionCoordinator; import com.baiyue.lastleg.security.OverseasPermissions; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; @@ -32,6 +33,7 @@ import org.springframework.web.server.ResponseStatusException; @Tag(name = "尾程入仓计划") public class InboundPlanController { @Resource private InboundPlanService service; + @Resource private ContainerInboundPlanConversionCoordinator containerConversion; @Resource private InboundFbaAppointmentService appointmentService; /** 按入仓计划明细分页查询可预约的入库订单行。 */ @@ -65,11 +67,12 @@ public class InboundPlanController { @Operation(summary = "按整柜批量生成剩余入库计划", description = "平台按 sourceContainerBookingId 锁定整柜订单；保留已有有效计划，只为剩余入仓码分别生成独立计划，整批原子且可幂等重放。Required authority: " + OverseasPermissions.INBOUND_ORDER_CREATE) public ApiResponse> convertContainer( @AuthenticationPrincipal AuthenticatedUser user, + @RequestHeader("Idempotency-Key") String idempotencyKey, @Valid @RequestBody ConvertContainerInboundPlanRequest request) { if (user == null || user.tenantId() == null || user.tenantId() != 0L || !"PLATFORM".equals(user.appCode())) { throw new ResponseStatusException(HttpStatus.FORBIDDEN, "仅平台账号可按整柜生成入库计划"); } - return ApiResponse.success(service.convertContainer(request.sourceContainerBookingId())); + return ApiResponse.success(containerConversion.confirm(request, idempotencyKey)); } @GetMapping("/container-preview") diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/dto/ConvertContainerInboundPlanRequest.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/dto/ConvertContainerInboundPlanRequest.java index b0908fdda..99dfc9c1d 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/dto/ConvertContainerInboundPlanRequest.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/dto/ConvertContainerInboundPlanRequest.java @@ -1,10 +1,14 @@ package com.baiyue.lastleg.inboundplan.dto; import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.Min; +import jakarta.validation.constraints.Pattern; import jakarta.validation.constraints.Size; /** 平台按来源头程柜预订身份批量生成入库计划。 */ public record ConvertContainerInboundPlanRequest( - @NotBlank @Size(max = 128) String sourceContainerBookingId + @NotBlank @Size(max = 128) String sourceContainerBookingId, + @Min(0) long snapshotVersion, + @NotBlank @Pattern(regexp = "^[0-9a-f]{64}$") String snapshotHash ) { } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/entity/ContainerPlanConversionCommand.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/entity/ContainerPlanConversionCommand.java new file mode 100644 index 000000000..327338d62 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/entity/ContainerPlanConversionCommand.java @@ -0,0 +1,27 @@ +package com.baiyue.lastleg.inboundplan.entity; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; + +import java.time.LocalDateTime; + +@Data +@TableName("ll_container_plan_conversion_command") +public class ContainerPlanConversionCommand { + @TableId(type = IdType.INPUT) private Long id; + private String sourceContainerBookingId; + private String idempotencyKey; + private String requestHash; + private Long snapshotVersion; + private String snapshotHash; + private Long warehouseId; + private String batchKey; + private String status; + private String preparationJson; + private String resultJson; + private String lastError; + private LocalDateTime createdAt; + private LocalDateTime updatedAt; +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/mapper/ContainerPlanConversionCommandMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/mapper/ContainerPlanConversionCommandMapper.java new file mode 100644 index 000000000..8ddca8d0b --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/mapper/ContainerPlanConversionCommandMapper.java @@ -0,0 +1,33 @@ +package com.baiyue.lastleg.inboundplan.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.baiyue.lastleg.inboundplan.entity.ContainerPlanConversionCommand; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; +import org.apache.ibatis.annotations.Update; + +import java.time.LocalDateTime; + +@Mapper +public interface ContainerPlanConversionCommandMapper extends BaseMapper { + @Select("SELECT * FROM ll_container_plan_conversion_command WHERE source_container_booking_id=#{bookingId} AND idempotency_key=#{key} LIMIT 1") + ContainerPlanConversionCommand find(@Param("bookingId") String bookingId, @Param("key") String key); + + @Select("SELECT * FROM ll_container_plan_conversion_command WHERE source_container_booking_id=#{bookingId} LIMIT 1") + ContainerPlanConversionCommand findByBooking(@Param("bookingId") String bookingId); + + @Update("UPDATE ll_container_plan_conversion_command SET status=#{status},result_json=COALESCE(#{resultJson},result_json),last_error=#{lastError},updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}") + int updateState(@Param("id") long id, @Param("expectedStatus") String expectedStatus, + @Param("status") String status, + @Param("resultJson") String resultJson, @Param("lastError") String lastError, + @Param("updatedAt") LocalDateTime updatedAt); + + @Update("UPDATE ll_container_plan_conversion_command SET status=#{nextStatus},last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status=#{expectedStatus}") + int claim(@Param("id") long id, @Param("expectedStatus") String expectedStatus, + @Param("nextStatus") String nextStatus, @Param("updatedAt") LocalDateTime updatedAt); + + @Update("UPDATE ll_container_plan_conversion_command SET last_error=NULL,updated_at=#{updatedAt} WHERE id=#{id} AND status='APPLYING' AND updated_at=#{expectedUpdatedAt}") + int reclaimStale(@Param("id") long id, @Param("expectedUpdatedAt") LocalDateTime expectedUpdatedAt, + @Param("updatedAt") LocalDateTime updatedAt); +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java index a5683449d..2b5043237 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGateway.java @@ -230,6 +230,39 @@ public class HttpWarehouseResourceGateway implements WarehouseResourceGateway { return response.data(); } + @Override + public PalletBatchResponse ensurePalletBatch(PalletBatchEnsure request) { + return palletBatch("/internal/v1/warehouse-pallet-batches/ensure-reserved", request); + } + + @Override + public PalletBatchResponse confirmPalletBatch(PalletBatchCommand request) { + return palletBatch("/internal/v1/warehouse-pallet-batches/confirm", request); + } + + @Override + public PalletBatchResponse compensatePalletBatch(PalletBatchCompensation request) { + return palletBatch("/internal/v1/warehouse-pallet-batches/compensate", request); + } + + private PalletBatchResponse palletBatch(String path, Object body) { + PalletBatchEnvelope response = client.post().uri(path) + .header("X-Internal-Service-Token", internalToken).body(body).retrieve() + .onStatus(HttpStatusCode::isError, (request, result) -> { + throw new ResponseStatusException(result.getStatusCode().is5xxServerError() + ? HttpStatus.SERVICE_UNAVAILABLE : HttpStatus.BAD_REQUEST, + "Warehouse pallet batch request failed"); + }).body(PalletBatchEnvelope.class); + if (response == null || !"200".equals(response.code()) || response.data() == null) { + HttpStatus status = response != null && "A0409".equals(response.code()) ? HttpStatus.CONFLICT + : response != null && "A0404".equals(response.code()) ? HttpStatus.NOT_FOUND + : HttpStatus.SERVICE_UNAVAILABLE; + throw new ResponseStatusException(status, response == null || response.message() == null + ? "Warehouse pallet batch returned no usable response" : response.message()); + } + return response.data(); + } + private record WarehousePalletResponse(String code, String message, WarehousePalletData data) {} private record WarehousePalletData(@JsonAlias("palletId") long id, String palletCode, long warehouseId, long locationId, String status, String fbaWarehouseCode, java.math.BigDecimal unitPrice, int version, @@ -239,4 +272,5 @@ public class HttpWarehouseResourceGateway implements WarehouseResourceGateway { private record ReserveData(long lockId) {} private record WarehouseReferenceResponse(String code, String message, WarehouseReference data) {} private record WarehouseLocationResponse(String code, String message, WarehouseLocationReference data) {} + private record PalletBatchEnvelope(String code, String message, PalletBatchResponse data) {} } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java index af7fd8eac..ca8a41da2 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/gateway/WarehouseResourceGateway.java @@ -1,5 +1,7 @@ package com.baiyue.lastleg.inboundplan.pda.gateway; +import java.util.List; + public interface WarehouseResourceGateway { WarehousePalletReference ensureDropshipPallet(String customerId, long warehouseId, String businessId); long reserve(String customerId, long warehouseId, long palletId, long locationId, @@ -18,6 +20,15 @@ public interface WarehouseResourceGateway { default WarehouseReference requireActiveSharedWarehouse(long warehouseId) { throw new UnsupportedOperationException("Shared warehouse resolution is not configured"); } + default PalletBatchResponse ensurePalletBatch(PalletBatchEnsure request) { + throw new UnsupportedOperationException("Warehouse pallet batch reservation is not configured"); + } + default PalletBatchResponse confirmPalletBatch(PalletBatchCommand request) { + throw new UnsupportedOperationException("Warehouse pallet batch confirmation is not configured"); + } + default PalletBatchResponse compensatePalletBatch(PalletBatchCompensation request) { + throw new UnsupportedOperationException("Warehouse pallet batch compensation is not configured"); + } record WarehousePalletReference(long palletId, String palletNo, long warehouseId, long locationId, String status, String fbaWarehouseCode, java.math.BigDecimal unitPrice, int version) { @@ -35,4 +46,14 @@ public interface WarehouseResourceGateway { String status) { } record WarehouseLocationReference(long locationId, String locationCode, String locationType, java.math.BigDecimal maxLoadKg, String status) { } + record PalletBatchEnsure(long warehouseId, String batchKey, List items) { } + record PalletBatchItemRequest(long customerId, String businessKey, String inboundPlanNo, + int sequenceNo, String palletType, String fbaWarehouseCode) { } + record PalletBatchCommand(long warehouseId, String batchKey, String commandKey) { } + record PalletBatchCompensation(long warehouseId, String batchKey, String commandKey, String reason) { } + record PalletBatchResponse(long batchId, String batchKey, String status, int attemptNo, + boolean replayed, List items) { } + record PalletBatchItem(long customerId, String businessKey, String inboundPlanNo, int sequenceNo, + long palletId, String palletCode, long locationId, long resourceLockId, + String status) { } } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinator.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinator.java new file mode 100644 index 000000000..5ff55ce38 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinator.java @@ -0,0 +1,289 @@ +package com.baiyue.lastleg.inboundplan.service; + +import com.baomidou.mybatisplus.core.toolkit.IdWorker; +import com.baiyue.lastleg.inboundplan.dto.ContainerInboundPlanPreviewResponse; +import com.baiyue.lastleg.inboundplan.dto.ConvertContainerInboundPlanRequest; +import com.baiyue.lastleg.inboundplan.dto.InboundPlanResponse; +import com.baiyue.lastleg.inboundplan.entity.ContainerPlanConversionCommand; +import com.baiyue.lastleg.inboundplan.mapper.ContainerPlanConversionCommandMapper; +import com.baiyue.lastleg.inboundplan.pda.gateway.WarehouseResourceGateway; +import com.baiyue.lastleg.service.LastlegOrderNoGenerator; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.annotation.Resource; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.http.HttpStatus; +import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.web.server.ResponseStatusException; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HexFormat; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; + +@Service +public class ContainerInboundPlanConversionCoordinator { + private static final java.time.Duration APPLYING_LEASE = java.time.Duration.ofMinutes(2); + @Resource private InboundPlanService plans; + @Resource private WarehouseResourceGateway warehouse; + @Resource private ContainerPlanConversionCommandMapper commands; + @Resource private LastlegOrderNoGenerator numbers; + @Resource private ObjectMapper objectMapper; + @Resource private PlatformTransactionManager transactionManager; + + public List confirm(ConvertContainerInboundPlanRequest request, String idempotencyKey) { + String bookingId = required(request.sourceContainerBookingId(), 128, "来源柜预订 ID 不能为空"); + String key = required(idempotencyKey, 128, "Idempotency-Key 必须为1至128个字符"); + String requestHash = hash(List.of(bookingId, request.snapshotVersion(), request.snapshotHash())); + ContainerPlanConversionCommand resolved = commands.find(bookingId, key); + if (resolved == null) resolved = commands.findByBooking(bookingId); + if (resolved == null) resolved = createCommand(bookingId, key, request, requestHash); + final ContainerPlanConversionCommand command = resolved; + requireSameRequest(command, requestHash); + Preparation preparation = read(command.getPreparationJson(), Preparation.class); + + if ("COMPLETED".equals(command.getStatus())) return result(command); + if ("APPLYING".equals(command.getStatus())) { + LocalDateTime now = LocalDateTime.now(); + if (command.getUpdatedAt() == null || command.getUpdatedAt().plus(APPLYING_LEASE).isAfter(now) + || commands.reclaimStale(command.getId(), command.getUpdatedAt(), now) != 1) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在生成入库计划，请勿重复提交"); + } + command.setUpdatedAt(now); + } + if (!"LOCAL_COMMITTED".equals(command.getStatus())) { + if (commands.claim(command.getId(), "PREPARED", "APPLYING", LocalDateTime.now()) != 1) { + ContainerPlanConversionCommand current = commands.findByBooking(bookingId); + if (current != null && "COMPLETED".equals(current.getStatus())) return result(current); + throw new ResponseStatusException(HttpStatus.CONFLICT, "该柜正在生成入库计划，请勿重复提交"); + } + command.setStatus("APPLYING"); + } + if (preparation.items().isEmpty()) { + try { + List existing = inTransaction(() -> { + List converted = plans.convertContainerReserved(bookingId, + request.snapshotVersion(), request.snapshotHash(), List.of()); + state(command.getId(), "APPLYING", "COMPLETED", write(converted), null); + return converted; + }); + command.setStatus("COMPLETED"); + return existing; + } catch (RuntimeException failure) { + resetForRetry(command, failure); + throw failure; + } + } + + WarehouseResourceGateway.PalletBatchEnsure ensure = new WarehouseResourceGateway.PalletBatchEnsure( + preparation.warehouseId(), command.getBatchKey(), preparation.items().stream().map(item -> + new WarehouseResourceGateway.PalletBatchItemRequest(Long.parseLong(item.customerId()), item.businessKey(), + item.planNo(), item.globalSequenceNo(), "STANDARD", item.fbaWarehouseCode())).toList()); + WarehouseResourceGateway.PalletBatchResponse reservation; + try { + reservation = warehouse.ensurePalletBatch(ensure); + requireBatch(reservation, command.getBatchKey(), preparation.items().size(), "RESERVED", "CONFIRMED"); + } catch (RuntimeException failure) { + resetForRetry(command, failure); + throw failure; + } + + List localResult; + if ("LOCAL_COMMITTED".equals(command.getStatus())) { + localResult = result(command); + } else { + try { + List preparedPlans = bind(preparation, reservation); + localResult = inTransaction(() -> { + List created = plans.convertContainerReserved(bookingId, + request.snapshotVersion(), request.snapshotHash(), preparedPlans); + state(command.getId(), "APPLYING", "LOCAL_COMMITTED", write(created), null); + return created; + }); + command.setStatus("LOCAL_COMMITTED"); + } catch (RuntimeException failure) { + try { + warehouse.compensatePalletBatch(new WarehouseResourceGateway.PalletBatchCompensation( + preparation.warehouseId(), command.getBatchKey(), "COMPENSATE:" + command.getId(), + "Lastleg local transaction did not commit")); + } catch (RuntimeException compensationFailure) { + failure.addSuppressed(compensationFailure); + } + resetForRetry(command, failure); + throw failure; + } + } + + WarehouseResourceGateway.PalletBatchResponse confirmed = warehouse.confirmPalletBatch( + new WarehouseResourceGateway.PalletBatchCommand(preparation.warehouseId(), command.getBatchKey(), + "CONFIRM:" + command.getId())); + requireBatch(confirmed, command.getBatchKey(), preparation.items().size(), "CONFIRMED"); + complete(command, localResult); + return localResult; + } + + private ContainerPlanConversionCommand createCommand(String bookingId, String key, + ConvertContainerInboundPlanRequest request, + String requestHash) { + ContainerInboundPlanPreviewResponse preview = plans.previewContainer(bookingId); + if (preview.snapshotVersion() != request.snapshotVersion() + || !Objects.equals(preview.snapshotHash(), request.snapshotHash()) || !preview.ready()) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级预览已变化或存在阻断，请刷新预览后重试"); + } + long warehouseId; + try { warehouseId = Long.parseLong(preview.overseasWarehouseId()); } + catch (RuntimeException exception) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级接柜仓 ID 无效", exception); + } + long commandId = IdWorker.getId(); + List items = new ArrayList<>(); + int globalSequence = 0; + for (ContainerInboundPlanPreviewResponse.Order order : preview.groups().stream() + .flatMap(group -> group.orders().stream()).filter(item -> "READY".equals(item.readiness())).toList()) { + long planId = IdWorker.getId(); + String planNo = numbers.next("INBOUND_PLAN", order.destinationWarehouseCode(), planId); + for (int localSequence = 1; localSequence <= order.suggestedPalletCount(); localSequence++) { + items.add(new PreparedItem(order.customerId(), Long.parseLong(order.orderId()), planId, planNo, + localSequence, ++globalSequence, "CIP:" + commandId + ":" + globalSequence, + "FBA".equals(order.destinationType()) ? order.destinationWarehouseCode() : null)); + } + } + Preparation preparation = new Preparation(warehouseId, List.copyOf(items)); + ContainerPlanConversionCommand command = new ContainerPlanConversionCommand(); + command.setId(commandId); command.setSourceContainerBookingId(bookingId); command.setIdempotencyKey(key); + command.setRequestHash(requestHash); command.setSnapshotVersion(request.snapshotVersion()); + command.setSnapshotHash(request.snapshotHash()); command.setWarehouseId(warehouseId); + command.setBatchKey("CIP:" + hash(bookingId) + ":" + commandId); command.setStatus("PREPARED"); + command.setPreparationJson(write(preparation)); command.setCreatedAt(LocalDateTime.now()); + command.setUpdatedAt(command.getCreatedAt()); + try { inTransaction(() -> { commands.insert(command); return null; }); } + catch (DuplicateKeyException concurrent) { + ContainerPlanConversionCommand existing = commands.findByBooking(bookingId); + if (existing == null) throw concurrent; + return existing; + } + return command; + } + + private List bind(Preparation preparation, + WarehouseResourceGateway.PalletBatchResponse response) { + Map byKey = new LinkedHashMap<>(); + for (WarehouseResourceGateway.PalletBatchItem item : response.items()) { + if (item.businessKey() == null || byKey.putIfAbsent(item.businessKey(), item) != null) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次包含重复业务键"); + } + } + if (response.items().stream().map(WarehouseResourceGateway.PalletBatchItem::palletId).distinct().count() != response.items().size() + || response.items().stream().map(WarehouseResourceGateway.PalletBatchItem::resourceLockId).distinct().count() != response.items().size()) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次包含重复托盘或资源锁"); + } + return preparation.items().stream().collect(Collectors.groupingBy(PreparedItem::orderId)).values().stream() + .map(items -> { + PreparedItem first = items.get(0); + return new InboundPlanService.PreparedPlan(first.planId(), first.planNo(), first.customerId(), + first.orderId(), items.stream().sorted(Comparator.comparingInt(PreparedItem::localSequenceNo)) + .map(item -> new InboundPlanService.PreparedPallet(item.localSequenceNo(), item.businessKey(), + requiredReservation(byKey.get(item.businessKey()), item))).toList()); + }).toList(); + } + + private WarehouseResourceGateway.PalletBatchItem requiredReservation( + WarehouseResourceGateway.PalletBatchItem value, PreparedItem expected) { + if (value == null || value.customerId() != Long.parseLong(expected.customerId()) + || !Objects.equals(value.businessKey(), expected.businessKey()) + || !Objects.equals(value.inboundPlanNo(), expected.planNo()) + || value.sequenceNo() != expected.globalSequenceNo() || value.palletId() <= 0 + || value.locationId() <= 0 || value.resourceLockId() <= 0 + || value.palletCode() == null || value.palletCode().isBlank()) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘预留响应与计划快照不一致"); + } + return value; + } + + private void requireBatch(WarehouseResourceGateway.PalletBatchResponse response, String batchKey, int size, + String... acceptedStatuses) { + if (response == null || !Objects.equals(batchKey, response.batchKey()) || response.items() == null + || response.items().size() != size || java.util.Arrays.stream(acceptedStatuses) + .noneMatch(status -> status.equals(response.status()))) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓托盘批次响应不完整或状态不一致"); + } + } + + private void requireSameRequest(ContainerPlanConversionCommand command, String requestHash) { + if (!MessageDigest.isEqual(command.getRequestHash().getBytes(StandardCharsets.UTF_8), + requestHash.getBytes(StandardCharsets.UTF_8))) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "同一幂等键不能用于不同柜级确认请求"); + } + } + + private List result(ContainerPlanConversionCommand command) { + if (command.getResultJson() == null) { + throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "柜级转换结果尚未可用"); + } + return read(command.getResultJson(), new TypeReference>() { }); + } + + private void complete(ContainerPlanConversionCommand command, List result) { + inTransaction(() -> { state(command.getId(), "LOCAL_COMMITTED", "COMPLETED", write(result), null); return null; }); + command.setStatus("COMPLETED"); + } + + private void resetForRetry(ContainerPlanConversionCommand command, RuntimeException failure) { + if ("APPLYING".equals(command.getStatus())) { + state(command.getId(), "APPLYING", "PREPARED", null, safeMessage(failure)); + command.setStatus("PREPARED"); + } + } + + private void state(long id, String expectedStatus, String status, String resultJson, String lastError) { + if (commands.updateState(id, expectedStatus, status, resultJson, lastError, LocalDateTime.now()) != 1) { + throw new IllegalStateException("无法更新柜级转换命令状态"); + } + } + + private T inTransaction(java.util.function.Supplier action) { + return new TransactionTemplate(transactionManager).execute(status -> action.get()); + } + + private String hash(Object value) { + try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(write(value).getBytes(StandardCharsets.UTF_8))); } + catch (NoSuchAlgorithmException exception) { throw new IllegalStateException(exception); } + } + private String write(Object value) { + try { return objectMapper.writeValueAsString(value); } + catch (JsonProcessingException exception) { throw new IllegalStateException("无法序列化柜级转换快照", exception); } + } + private T read(String value, Class type) { + try { return objectMapper.readValue(value, type); } + catch (JsonProcessingException exception) { throw new IllegalStateException("无法读取柜级转换快照", exception); } + } + private T read(String value, TypeReference type) { + try { return objectMapper.readValue(value, type); } + catch (JsonProcessingException exception) { throw new IllegalStateException("无法读取柜级转换结果", exception); } + } + private static String required(String value, int max, String message) { + String normalized = value == null || value.isBlank() ? null : value.trim(); + if (normalized == null || normalized.length() > max) throw new ResponseStatusException(HttpStatus.BAD_REQUEST, message); + return normalized; + } + private static String safeMessage(Throwable value) { + String message = value.getMessage(); + return message == null ? value.getClass().getSimpleName() : message.substring(0, Math.min(1000, message.length())); + } + + public record Preparation(long warehouseId, List items) { } + public record PreparedItem(String customerId, long orderId, long planId, String planNo, + int localSequenceNo, int globalSequenceNo, String businessKey, + String fbaWarehouseCode) { } +} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundPlanService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundPlanService.java index 058a8a0dd..b2ab343e1 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundPlanService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/service/InboundPlanService.java @@ -231,6 +231,7 @@ public class InboundPlanService { || row.packageCount() > 0 && row.volumeCbm().signum() == 0) blockers.add("CARGO_FACT_INVALID"); } Integer suggested = suggestedPalletCount(row, blockers); + if (!planned && (suggested == null || suggested <= 0)) blockers.add("PALLET_COUNT_NOT_READY"); String destinationType = destinationType(row); String destinationCode = "FBA".equals(destinationType) && trimToNull(row.fbaWarehouseCode()) != null ? row.fbaWarehouseCode().trim() : row.destinationWarehouseCode(); @@ -335,11 +336,23 @@ public class InboundPlanService { List orders, List sourcePalletNos, List wholePallets, ManualInboundPlanRequest manual, WarehouseResourceGateway.WarehouseReference receivingWarehouse) { + return createPlan(customerId, selectedOrderId, orders, sourcePalletNos, wholePallets, manual, + receivingWarehouse, null, Map.of()); + } + + private InboundPlanResponse createPlan(String customerId, Long selectedOrderId, + List orders, List sourcePalletNos, + List wholePallets, ManualInboundPlanRequest manual, + WarehouseResourceGateway.WarehouseReference receivingWarehouse, + PreparedPlan preparedPlan, + Map reservedPallets) { String containerNo = requiredContainerNo(orders.get(0).containerNo()); LocalDateTime now = LocalDateTime.now(); InboundPlan plan = new InboundPlan(); - plan.setId(IdWorker.getId()); - plan.setPlanNo(orderNoGenerator.next("INBOUND_PLAN", orders.get(0).destinationWarehouseCode(), plan.getId())); + plan.setId(preparedPlan == null ? IdWorker.getId() : preparedPlan.planId()); + plan.setPlanNo(preparedPlan == null + ? orderNoGenerator.next("INBOUND_PLAN", orders.get(0).destinationWarehouseCode(), plan.getId()) + : preparedPlan.planNo()); plan.setCustomerId(customerId); plan.setSourceOrderId(selectedOrderId); plan.setContainerNo(containerNo); @@ -384,14 +397,14 @@ public class InboundPlanService { palletCounts = Map.of(orders.get(0).orderId(), manual.palletCount()); } else if (wholePallets == null) { Map persistedPallets = createPallets(plan, allocation, - orderById, sourcePalletNos == null ? List.of() : sourcePalletNos); + orderById, sourcePalletNos == null ? List.of() : sourcePalletNos, reservedPallets); createAllocations(plan, allocation, lineByOrder, persistedPallets); palletCounts = allocation.allocations().stream().collect(Collectors.groupingBy( PalletAllocator.AllocationDraft::orderId, Collectors.collectingAndThen(Collectors.mapping( item -> item.pallet().sequenceNo(), Collectors.toSet()), java.util.Set::size))); } else { SourceOrderSnapshot order = orders.get(0); - createWholePallets(plan, order, lineByOrder.get(order.orderId()), wholePallets); + createWholePallets(plan, order, lineByOrder.get(order.orderId()), wholePallets, reservedPallets); palletCounts = Map.of(order.orderId(), wholePallets.size()); } for (SourceOrderSnapshot order : orders) { @@ -417,10 +430,32 @@ public class InboundPlanService { /** 同一来源柜的每个尾程订单独立建计划；整个柜级批次共用一个事务。 */ @Transactional public List convertContainer(String sourceContainerBookingId) { + return convertContainerInternal(sourceContainerBookingId, null, null, Map.of()); + } + + @Transactional + public List convertContainerReserved(String sourceContainerBookingId, + long snapshotVersion, String snapshotHash, + List preparedPlans) { + Map prepared = preparedPlans.stream().collect(Collectors.toMap( + PreparedPlan::orderId, value -> value)); + return convertContainerInternal(sourceContainerBookingId, snapshotVersion, snapshotHash, prepared); + } + + private List convertContainerInternal(String sourceContainerBookingId, + Long snapshotVersion, String snapshotHash, + Map preparedPlans) { String bookingId = trimToNull(sourceContainerBookingId); if (bookingId == null) throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "来源柜预订 ID 不能为空"); List refs = orderMapper.lockContainerOrders(bookingId); if (refs.isEmpty()) throw new ResponseStatusException(HttpStatus.NOT_FOUND, "来源柜没有尾程订单"); + if (snapshotVersion != null) { + ContainerInboundPlanPreviewResponse current = previewContainer(bookingId); + if (current.snapshotVersion() != snapshotVersion || !Objects.equals(current.snapshotHash(), snapshotHash) + || !current.ready()) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级预览已变化或存在阻断，请刷新预览后重试"); + } + } String containerNo = null; for (InboundPlanOrderMapper.ContainerOrderRef ref : refs) { String current = requiredContainerNo(ref.containerNo()); @@ -481,8 +516,15 @@ public class InboundPlanService { for (int pendingIndex = 0; pendingIndex < pendingIndexes.size(); pendingIndex++) { int refIndex = pendingIndexes.get(pendingIndex); InboundPlanOrderMapper.ContainerOrderRef ref = refs.get(refIndex); + PreparedPlan prepared = preparedPlans.get(ref.orderId()); + if (snapshotVersion != null && (prepared == null || !Objects.equals(prepared.customerId(), ref.customerId()))) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "柜级托盘预留与待生成计划不一致"); + } + Map reserved = prepared == null ? Map.of() + : prepared.pallets().stream().collect(Collectors.toMap(PreparedPallet::sequenceNo, + PreparedPallet::reservation)); created.put(refIndex, createPlan(ref.customerId(), ref.orderId(), snapshots.get(pendingIndex), List.of(), - wholePallets.get(pendingIndex), null, null)); + wholePallets.get(pendingIndex), null, null, prepared, reserved)); } List result = new ArrayList<>(refs.size()); for (int index = 0; index < refs.size(); index++) { @@ -492,6 +534,11 @@ public class InboundPlanService { return List.copyOf(result); } + public record PreparedPlan(long planId, String planNo, String customerId, long orderId, + List pallets) { } + public record PreparedPallet(int sequenceNo, String businessKey, + WarehouseResourceGateway.PalletBatchItem reservation) { } + private WholePalletInput wholePallet(DropshipPalletSnapshot pallet) { return new WholePalletInput(pallet.sourcePalletNo(), Math.toIntExact(pallet.packageCount())); } @@ -569,7 +616,8 @@ public class InboundPlanService { } private void createWholePallets(InboundPlan plan, SourceOrderSnapshot order, InboundPlanLine line, - List sourcePallets) { + List sourcePallets, + Map reservedPallets) { BigDecimal remainingVolume = volume(order); int remainingPackages = order.packageCount(); for (int index = 0; index < sourcePallets.size(); index++) { @@ -586,6 +634,7 @@ public class InboundPlanService { pallet.setDestinationWarehouseName(order.destinationWarehouseName()); pallet.setCapacityCbm(PalletAllocator.PALLET_CAPACITY_CBM); pallet.setAllocatedVolumeCbm(palletVolume); pallet.setSequenceNo(index + 1); pallet.setAssignmentStatus("UNASSIGNED"); + applyReservedPallet(pallet, reservedPallets.get(index + 1)); pallet.setActualVolumeCbm(BigDecimal.ZERO); pallet.setActualPackageCount(0); palletMapper.insert(pallet); InboundPlanPalletAllocation allocation = new InboundPlanPalletAllocation(); @@ -707,7 +756,7 @@ public class InboundPlanService { private Map createPallets( InboundPlan plan, PalletAllocator.Result allocation, Map orderById, - List sourcePalletNos) { + List sourcePalletNos, Map reservedPallets) { Map warehouseSamples = orderById.values().stream() .collect(Collectors.toMap(this::warehouseKey, value -> value, (left, right) -> left)); Map result = new HashMap<>(); @@ -729,6 +778,7 @@ public class InboundPlanService { pallet.setAllocatedVolumeCbm(draft.allocatedVolumeCbm()); pallet.setSequenceNo(result.size() + 1); pallet.setAssignmentStatus("UNASSIGNED"); + applyReservedPallet(pallet, reservedPallets.get(pallet.getSequenceNo())); pallet.setActualVolumeCbm(BigDecimal.ZERO); pallet.setActualPackageCount(0); palletMapper.insert(pallet); @@ -737,6 +787,20 @@ public class InboundPlanService { return result; } + private void applyReservedPallet(InboundPlanPallet pallet, + WarehouseResourceGateway.PalletBatchItem reserved) { + if (reserved == null) return; + if (trimToNull(reserved.palletCode()) == null || reserved.palletCode().length() > 128) { + throw new ResponseStatusException(HttpStatus.CONFLICT, "海外仓返回的真实托盘码无效"); + } + pallet.setPalletNo(reserved.palletCode()); + pallet.setWarehousePalletId(reserved.palletId()); + pallet.setWarehouseLocationId(reserved.locationId()); + pallet.setWarehouseResourceLockId(reserved.resourceLockId()); + pallet.setAssignmentStatus("RESERVED"); + pallet.setLockedAt(LocalDateTime.now()); + } + private void createAllocations(InboundPlan plan, PalletAllocator.Result allocation, Map lineByOrder, Map pallets) { diff --git a/lastleg-service/src/main/resources/db/lastleg.sql b/lastleg-service/src/main/resources/db/lastleg.sql index 41a57fe57..49374729d 100644 --- a/lastleg-service/src/main/resources/db/lastleg.sql +++ b/lastleg-service/src/main/resources/db/lastleg.sql @@ -578,6 +578,27 @@ CREATE TABLE IF NOT EXISTS ll_inbound_plan ( KEY idx_ll_inbound_plan_container (customer_id, container_no) ) ENGINE=InnoDB COMMENT='尾程入库计划表：每张来源入仓单生成一张入库计划，同柜计划共享集装箱号'; +CREATE TABLE IF NOT EXISTS ll_container_plan_conversion_command ( + id BIGINT NOT NULL COMMENT '柜级转换命令ID', + source_container_booking_id VARCHAR(128) NOT NULL COMMENT '稳定来源柜预订ID', + idempotency_key VARCHAR(128) NOT NULL COMMENT '调用方幂等键', + request_hash CHAR(64) NOT NULL COMMENT '确认请求SHA-256摘要', + snapshot_version BIGINT NOT NULL COMMENT '确认的柜级快照版本', + snapshot_hash CHAR(64) NOT NULL COMMENT '确认的柜级快照SHA-256', + warehouse_id BIGINT NOT NULL COMMENT '海外接柜仓ID', + batch_key VARCHAR(160) NOT NULL COMMENT '海外仓托盘批次幂等键', + status VARCHAR(32) NOT NULL COMMENT 'PREPARED/APPLYING/LOCAL_COMMITTED/COMPLETED', + preparation_json JSON NOT NULL COMMENT '稳定计划号与托盘业务键快照', + result_json JSON NULL COMMENT '本地计划提交结果', + last_error VARCHAR(1000) NULL COMMENT '最近一次跨服务错误', + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY(id), + UNIQUE KEY uk_ll_container_conversion_command(source_container_booking_id), + UNIQUE KEY uk_ll_container_conversion_batch(warehouse_id,batch_key), + KEY idx_ll_container_conversion_status(status,updated_at) +) ENGINE=InnoDB COMMENT='柜级入库计划转换Saga与幂等命令'; + CREATE TABLE IF NOT EXISTS ll_inbound_plan_line ( id BIGINT NOT NULL COMMENT '入库计划明细ID（雪花ID）', customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识', @@ -616,7 +637,7 @@ CREATE TABLE IF NOT EXISTS ll_inbound_plan_pallet ( id BIGINT NOT NULL COMMENT '计划托盘ID（雪花ID）', customer_id VARCHAR(128) NOT NULL COMMENT '客户账号标识', plan_id BIGINT NOT NULL COMMENT '入库计划ID', - pallet_no VARCHAR(64) NOT NULL COMMENT '当前托盘编号：分配前为计划占位码，分配后为海外仓实物托盘码', + pallet_no VARCHAR(128) NOT NULL COMMENT '海外仓资源服务返回的真实托盘码', source_pallet_no VARCHAR(128) NULL COMMENT '一件代发不可变来源托盘号快照', pallet_type VARCHAR(32) NULL COMMENT '人工计划托盘类型', destination_warehouse_id VARCHAR(128) NULL COMMENT '目的仓ID快照', @@ -694,7 +715,7 @@ ALTER TABLE ll_inbound_plan_pallet COMMENT='尾程入库计划托盘表：按目 ALTER TABLE ll_inbound_plan_pallet MODIFY COLUMN id BIGINT NOT NULL COMMENT '计划托盘ID（雪花ID）', MODIFY COLUMN plan_id BIGINT NOT NULL COMMENT '入库计划ID', - MODIFY COLUMN pallet_no VARCHAR(64) NOT NULL COMMENT '当前托盘编号：分配前为计划占位码，分配后为海外仓实物托盘码', + MODIFY COLUMN pallet_no VARCHAR(128) NOT NULL COMMENT '海外仓资源服务返回的真实托盘码', MODIFY COLUMN destination_warehouse_id VARCHAR(128) NULL COMMENT '目的仓ID快照', MODIFY COLUMN destination_warehouse_code VARCHAR(128) NULL COMMENT '目的仓编码快照', MODIFY COLUMN destination_warehouse_name VARCHAR(255) NULL COMMENT '目的仓名称快照', diff --git a/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml b/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml index 87d998a50..7f185ee30 100644 --- a/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml +++ b/lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml @@ -433,17 +433,21 @@ paths: /api/platform/v1/last-leg/inbound-plans/convert-container: post: summary: 按整柜批量生成剩余入库计划 - description: 权限 platform:overseas-warehouse:inbound-order:create；使用稳定 sourceContainerBookingId 锁定整柜订单，并要求全柜尾程订单及既有有效计划的真实 containerNo 与接柜仓快照一致。已有有效计划保持不变，只为剩余尾程订单/入仓码分别生成独立计划；响应按柜内订单稳定顺序返回全部已有及新增计划。整批原子且幂等，剩余为0时直接返回已有结果；任何快照漂移均冲突且不部分写入。 + description: 权限 platform:overseas-warehouse:inbound-order:create；调用方必须先取柜级预览并原样回传 snapshotVersion/snapshotHash。服务端以 sourceContainerBookingId 建立唯一 Saga 命令，先向海外仓资源服务按票、按托盘批量预留真实 palletId/palletCode/locationId/resourceLockId，再在一个本地事务中锁整柜并为每票生成独立计划，最后确认资源批次。本地任一票失败时整柜回滚并补偿资源；本地已提交但远端确认失败时重放只重试确认。同柜不同 Idempotency-Key 也不会产生第二批资源。 security: [{ bearerAuth: [] }] + parameters: + - { name: Idempotency-Key, in: header, required: true, schema: { type: string, minLength: 1, maxLength: 128 } } requestBody: required: true content: application/json: schema: type: object - required: [sourceContainerBookingId] + required: [sourceContainerBookingId, snapshotVersion, snapshotHash] properties: sourceContainerBookingId: { type: string, minLength: 1, maxLength: 128 } + snapshotVersion: { type: integer, format: int64, minimum: 0 } + snapshotHash: { type: string, pattern: '^[0-9a-f]{64}$' } responses: { '200': { description: 统一 ApiResponse 包装的入库计划数组 } } /api/platform/v1/last-leg/inbound-plans/container-preview: get: diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanControllerTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanControllerTest.java index d33ad32c2..d29022f19 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanControllerTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanControllerTest.java @@ -8,6 +8,7 @@ import com.baiyue.lastleg.inboundplan.dto.ManualInboundPlanRequest; import com.baiyue.lastleg.inboundplan.dto.UpdateFbaAppointmentRequest; import com.baiyue.lastleg.inboundplan.service.InboundFbaAppointmentService; import com.baiyue.lastleg.inboundplan.service.InboundPlanService; +import com.baiyue.lastleg.inboundplan.service.ContainerInboundPlanConversionCoordinator; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -30,6 +31,7 @@ import static org.mockito.Mockito.when; class InboundPlanControllerTest { @Mock InboundPlanService service; @Mock InboundFbaAppointmentService appointmentService; + @Mock ContainerInboundPlanConversionCoordinator containerConversion; private InboundPlanController controller; @BeforeEach @@ -37,6 +39,7 @@ class InboundPlanControllerTest { controller = new InboundPlanController(); ReflectionTestUtils.setField(controller, "service", service); ReflectionTestUtils.setField(controller, "appointmentService", appointmentService); + ReflectionTestUtils.setField(controller, "containerConversion", containerConversion); } @Test @@ -79,19 +82,21 @@ class InboundPlanControllerTest { @Test void platformContainerConversionUsesStableBookingIdentity() { - controller.convertContainer(platform(), new ConvertContainerInboundPlanRequest(" BOOKING-1 ")); + var request = new ConvertContainerInboundPlanRequest(" BOOKING-1 ", 3, "a".repeat(64)); + controller.convertContainer(platform(), "convert-key", request); - verify(service).convertContainer(" BOOKING-1 "); + verify(containerConversion).confirm(request, "convert-key"); } @Test void merchantCannotConvertAWholeContainer() { AuthenticatedUser customer = new AuthenticatedUser(8L, 100L, 100L, 9L, "customer", "MERCHANT", "WEB", Set.of()); - assertThatThrownBy(() -> controller.convertContainer(customer, - new ConvertContainerInboundPlanRequest("BOOKING-1"))) + assertThatThrownBy(() -> controller.convertContainer(customer, "convert-key", + new ConvertContainerInboundPlanRequest("BOOKING-1", 3, "a".repeat(64)))) .isInstanceOf(ResponseStatusException.class); - verify(service, never()).convertContainer(org.mockito.ArgumentMatchers.any()); + verify(containerConversion, never()).confirm(org.mockito.ArgumentMatchers.any(), + org.mockito.ArgumentMatchers.any()); } private static AuthenticatedUser platform() { diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java index be03e6fa4..1da4230b5 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/gateway/HttpWarehouseResourceGatewayTest.java @@ -8,6 +8,8 @@ import org.springframework.test.web.client.MockRestServiceServer; import org.springframework.web.client.RestClient; import org.springframework.web.server.ResponseStatusException; +import java.util.List; + import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo; @@ -141,4 +143,38 @@ class HttpWarehouseResourceGatewayTest { server.verify(); } + @Test + void ensurePalletBatchAcceptsJacksonStringEncodedLongs() { + server.expect(requestTo("http://warehouse/internal/v1/warehouse-pallet-batches/ensure-reserved")) + .andRespond(withSuccess(""" + {"code":"200","data":{"batchId":"41","batchKey":"BATCH-1","status":"RESERVED", + "attemptNo":1,"replayed":false,"items":[{"customerId":"100","businessKey":"CIP:1:1", + "inboundPlanNo":"IP-21","sequenceNo":1,"palletId":"51","palletCode":"PLT-51", + "locationId":"61","resourceLockId":"71","status":"RESERVED"}]}} + """, MediaType.APPLICATION_JSON)); + + var result = gateway.ensurePalletBatch(new WarehouseResourceGateway.PalletBatchEnsure(31L, "BATCH-1", + List.of(new WarehouseResourceGateway.PalletBatchItemRequest( + 100L, "CIP:1:1", "IP-21", 1, "STANDARD", "ONT8")))); + + assertThat(result.batchId()).isEqualTo(41L); + assertThat(result.items().get(0).resourceLockId()).isEqualTo(71L); + server.verify(); + } + + @Test + void palletBatchMapsBusinessConflictEnvelopeEvenWithHttp200() { + server.expect(requestTo("http://warehouse/internal/v1/warehouse-pallet-batches/confirm")) + .andRespond(withSuccess(""" + {"code":"A0409","message":"Batch request hash conflict","data":null} + """, MediaType.APPLICATION_JSON)); + + assertThatThrownBy(() -> gateway.confirmPalletBatch( + new WarehouseResourceGateway.PalletBatchCommand(31L, "BATCH-1", "CONFIRM:1"))) + .isInstanceOf(ResponseStatusException.class) + .hasMessageContaining("409") + .hasMessageContaining("Batch request hash conflict"); + server.verify(); + } + } diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinatorTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinatorTest.java new file mode 100644 index 000000000..196833bd4 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/ContainerInboundPlanConversionCoordinatorTest.java @@ -0,0 +1,126 @@ +package com.baiyue.lastleg.inboundplan.service; + +import com.baiyue.lastleg.inboundplan.dto.ConvertContainerInboundPlanRequest; +import com.baiyue.lastleg.inboundplan.entity.ContainerPlanConversionCommand; +import com.baiyue.lastleg.inboundplan.mapper.ContainerPlanConversionCommandMapper; +import com.baiyue.lastleg.inboundplan.pda.gateway.WarehouseResourceGateway; +import com.fasterxml.jackson.databind.json.JsonMapper; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.web.server.ResponseStatusException; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class ContainerInboundPlanConversionCoordinatorTest { + private static final String BOOKING_ID = "BOOKING-1"; + private static final String SNAPSHOT_HASH = "a".repeat(64); + + @Mock InboundPlanService plans; + @Mock WarehouseResourceGateway warehouse; + @Mock ContainerPlanConversionCommandMapper commands; + + private ContainerInboundPlanConversionCoordinator coordinator; + private JsonMapper objectMapper; + + @BeforeEach + void setUp() { + coordinator = new ContainerInboundPlanConversionCoordinator(); + objectMapper = JsonMapper.builder().findAndAddModules().build(); + ReflectionTestUtils.setField(coordinator, "plans", plans); + ReflectionTestUtils.setField(coordinator, "warehouse", warehouse); + ReflectionTestUtils.setField(coordinator, "commands", commands); + ReflectionTestUtils.setField(coordinator, "objectMapper", objectMapper); + } + + @Test + void anotherIdempotencyKeyForCompletedCabinetReplaysWithoutCreatingAnotherBatch() throws Exception { + ConvertContainerInboundPlanRequest request = request(); + ContainerPlanConversionCommand existing = command("FIRST-KEY", "COMPLETED", request); + existing.setResultJson("[]"); + when(commands.find(BOOKING_ID, "SECOND-KEY")).thenReturn(null); + when(commands.findByBooking(BOOKING_ID)).thenReturn(existing); + + assertEquals(List.of(), coordinator.confirm(request, "SECOND-KEY")); + + verify(warehouse, never()).ensurePalletBatch(org.mockito.ArgumentMatchers.any()); + verify(commands, never()).insert(org.mockito.ArgumentMatchers.any(ContainerPlanConversionCommand.class)); + } + + @Test + void concurrentCabinetClaimRejectsFollowerBeforeWarehouseReservation() throws Exception { + ConvertContainerInboundPlanRequest request = request(); + ContainerPlanConversionCommand applying = command("FIRST-KEY", "APPLYING", request); + when(commands.find(BOOKING_ID, "SECOND-KEY")).thenReturn(null); + when(commands.findByBooking(BOOKING_ID)).thenReturn(applying); + + ResponseStatusException error = assertThrows(ResponseStatusException.class, + () -> coordinator.confirm(request, "SECOND-KEY")); + + assertEquals(409, error.getStatusCode().value()); + verify(warehouse, never()).ensurePalletBatch(org.mockito.ArgumentMatchers.any()); + } + + @Test + void failedConditionalClaimStopsBeforeWarehouseReservation() throws Exception { + ConvertContainerInboundPlanRequest request = request(); + ContainerPlanConversionCommand prepared = command("FIRST-KEY", "PREPARED", request); + when(commands.find(BOOKING_ID, "SECOND-KEY")).thenReturn(null); + when(commands.findByBooking(BOOKING_ID)).thenReturn(prepared); + when(commands.claim(org.mockito.ArgumentMatchers.eq(1L), org.mockito.ArgumentMatchers.eq("PREPARED"), + org.mockito.ArgumentMatchers.eq("APPLYING"), org.mockito.ArgumentMatchers.any())).thenReturn(0); + + ResponseStatusException error = assertThrows(ResponseStatusException.class, + () -> coordinator.confirm(request, "SECOND-KEY")); + + assertEquals(409, error.getStatusCode().value()); + verify(warehouse, never()).ensurePalletBatch(org.mockito.ArgumentMatchers.any()); + } + + @Test + void reservationMustMatchEveryStableItemIdentity() throws Exception { + var expected = new ContainerInboundPlanConversionCoordinator.PreparedItem( + "100", 11L, 21L, "IP-21", 1, 1, "CIP:1:1", "ONT8"); + var preparation = new ContainerInboundPlanConversionCoordinator.Preparation(31L, List.of(expected)); + var response = new WarehouseResourceGateway.PalletBatchResponse(41L, "BATCH-1", "RESERVED", 1, + false, List.of(new WarehouseResourceGateway.PalletBatchItem( + 100L, "CIP:1:1", "WRONG-PLAN", 1, 51L, "PLT-51", 61L, 71L, "RESERVED"))); + + ResponseStatusException error = assertThrows(ResponseStatusException.class, + () -> ReflectionTestUtils.invokeMethod(coordinator, "bind", preparation, response)); + + assertEquals(409, error.getStatusCode().value()); + } + + private ConvertContainerInboundPlanRequest request() { + return new ConvertContainerInboundPlanRequest(BOOKING_ID, 7L, SNAPSHOT_HASH); + } + + private ContainerPlanConversionCommand command(String key, String status, + ConvertContainerInboundPlanRequest request) throws Exception { + ContainerPlanConversionCommand command = new ContainerPlanConversionCommand(); + command.setId(1L); + command.setSourceContainerBookingId(BOOKING_ID); + command.setIdempotencyKey(key); + command.setRequestHash(ReflectionTestUtils.invokeMethod(coordinator, "hash", + List.of(BOOKING_ID, request.snapshotVersion(), request.snapshotHash()))); + command.setSnapshotVersion(request.snapshotVersion()); + command.setSnapshotHash(request.snapshotHash()); + command.setWarehouseId(31L); + command.setBatchKey("BATCH-1"); + command.setStatus(status); + command.setPreparationJson(objectMapper.writeValueAsString( + new ContainerInboundPlanConversionCoordinator.Preparation(31L, List.of()))); + return command; + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/InboundPlanServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/InboundPlanServiceTest.java index 84fc1f422..b2afab27b 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/InboundPlanServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/service/InboundPlanServiceTest.java @@ -424,6 +424,8 @@ class InboundPlanServiceTest { void containerConversionPrevalidatesEveryOrderBeforeWritingIndependentPlans() throws Exception { assertNotNull(InboundPlanService.class.getMethod("convertContainer", String.class) .getAnnotation(Transactional.class)); + assertNotNull(InboundPlanService.class.getMethod("convertContainerReserved", String.class, + long.class, String.class, List.class).getAnnotation(Transactional.class)); when(orderMapper.lockContainerOrders("BOOKING-1")).thenReturn(List.of( new InboundPlanOrderMapper.ContainerOrderRef("100", 1L, "PENDING_ASSIGNMENT", "MSCU6639870", "STANDARD"), new InboundPlanOrderMapper.ContainerOrderRef("200", 2L, "DESTINATION_ASSIGNED", "MSCU6639870", "STANDARD")));