Files
baiyue-firstleg/warehouse-overseas-service/src/main/java/com/baiyue/warehouse/service/WarehousePalletBatchService.java
T

350 lines
19 KiB
Java

package com.baiyue.warehouse.service;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baiyue.common.core.api.ResultCode;
import com.baiyue.common.core.exception.BizException;
import com.baiyue.warehouse.dto.PalletRequests;
import com.baiyue.warehouse.dto.WarehousePalletBatchRequests;
import com.baiyue.warehouse.dto.WarehousePalletBatchResponse;
import com.baiyue.warehouse.dto.WarehouseResourceLockRequests;
import com.baiyue.warehouse.dto.WarehouseResourceLockResponse;
import com.baiyue.warehouse.entity.WarehousePalletBatchEntity;
import com.baiyue.warehouse.entity.WarehousePalletBatchItemEntity;
import com.baiyue.warehouse.entity.WarehousePalletBindingEntity;
import com.baiyue.warehouse.entity.WarehousePalletEntity;
import com.baiyue.warehouse.mapper.WarehousePalletBatchItemMapper;
import com.baiyue.warehouse.mapper.WarehousePalletBatchMapper;
import com.baiyue.warehouse.repository.WarehouseResourceRepository;
import com.baiyue.warehouse.repository.WarehouseResourceLockRepository;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.math.BigDecimal;
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.HashSet;
import java.util.HexFormat;
import java.util.List;
import java.util.Objects;
import java.util.Set;
@Service
public class WarehousePalletBatchService {
private static final String BUSINESS_TYPE = "INBOUND_PLAN_AUTO";
@Resource private WarehousePalletBatchMapper batches;
@Resource private WarehousePalletBatchItemMapper items;
@Resource private WarehouseResourceService resources;
@Resource private WarehouseResourceLockService locks;
@Resource private WarehouseResourceLockRepository lockRecords;
@Resource private WarehouseResourceRepository repository;
@Resource private ObjectMapper objectMapper;
@Transactional
public WarehousePalletBatchResponse ensureReserved(WarehousePalletBatchRequests.EnsureReserved request) {
WarehousePalletBatchRequests.EnsureReserved normalized = normalize(request);
String requestHash = hash(normalized);
WarehousePalletBatchEntity batch = new WarehousePalletBatchEntity();
batch.setId(IdWorker.getId());
batch.setWarehouseId(normalized.warehouseId());
batch.setBatchKey(normalized.batchKey());
batch.setRequestHash(requestHash);
batch.setStatus("RESERVED");
batch.setAttemptNo(1);
batch.setCreatedAt(LocalDateTime.now());
batch.setUpdatedAt(batch.getCreatedAt());
if (batches.insertIgnore(batch) == 0) {
WarehousePalletBatchEntity concurrent = batches.lockByKey(normalized.warehouseId(), normalized.batchKey());
if (concurrent == null) {
throw conflict("Pallet batch was created concurrently but cannot be loaded");
}
requireSameRequest(concurrent, normalized.warehouseId(), requestHash);
if ("COMPENSATED".equals(concurrent.getStatus())) return reactivate(concurrent);
return response(concurrent, items.lockByBatch(concurrent.getId()), true);
}
for (WarehousePalletBatchRequests.Item requested : normalized.items()) {
if (repository.lockPalletBinding(requested.customerId(), normalized.warehouseId(), BUSINESS_TYPE,
requested.businessKey()) != null) {
throw conflict("Inbound plan pallet business key is already bound");
}
}
List<WarehousePalletBatchItemEntity> created = new ArrayList<>();
for (WarehousePalletBatchRequests.Item requested : normalized.items()) {
Long palletId = resources.createPallet(WarehouseResourceService.PUBLIC_RESOURCE_SCOPE,
normalized.warehouseId(), new PalletRequests.SaveRequest(null, requested.palletType(), null,
BigDecimal.ZERO, "IDLE", null, "Auto-created for inbound plan " + requested.inboundPlanNo(),
requested.fbaWarehouseCode()));
WarehousePalletEntity pallet = resources.pallet(WarehouseResourceService.PUBLIC_RESOURCE_SCOPE,
normalized.warehouseId(), palletId);
WarehouseResourceLockResponse lock = locks.reserve(new WarehouseResourceLockRequests.ReserveRequest(
requested.customerId(), normalized.warehouseId(), pallet.getId(), pallet.getLocationId(),
BUSINESS_TYPE, requested.businessKey(), reservationKey(batch, requested.sequenceNo())));
WarehousePalletBatchItemEntity item = new WarehousePalletBatchItemEntity();
item.setId(IdWorker.getId());
item.setBatchId(batch.getId());
item.setCustomerId(requested.customerId());
item.setWarehouseId(normalized.warehouseId());
item.setBusinessKey(requested.businessKey());
item.setInboundPlanNo(requested.inboundPlanNo());
item.setSequenceNo(requested.sequenceNo());
item.setPalletId(pallet.getId());
item.setPalletCode(pallet.getPalletCode());
item.setLocationId(pallet.getLocationId());
item.setResourceLockId(lock.lockId());
item.setStatus("RESERVED");
item.setCreatedAt(LocalDateTime.now());
item.setUpdatedAt(item.getCreatedAt());
items.insert(item);
created.add(item);
}
return response(batch, created, false);
}
@Transactional
public WarehousePalletBatchResponse confirm(WarehousePalletBatchRequests.Confirm request) {
WarehousePalletBatchRequests.Confirm normalized = normalize(request);
String commandHash = hash(normalized);
WarehousePalletBatchEntity batch = requiredBatch(normalized.warehouseId(), normalized.batchKey());
List<WarehousePalletBatchItemEntity> batchItems = items.lockByBatch(batch.getId());
if ("CONFIRMED".equals(batch.getStatus())) {
requireSameCommand(batch.getConfirmCommandHash(), commandHash);
return response(batch, batchItems, true);
}
if (!"RESERVED".equals(batch.getStatus())) {
throw conflict("Only a RESERVED pallet batch can be confirmed");
}
for (WarehousePalletBatchItemEntity item : batchItems) {
var lock = lockRecords.find(item.getCustomerId(), item.getResourceLockId());
if (lock == null || !"ACTIVE".equals(lock.getStatus())
|| !Objects.equals(lock.getWarehouseId(), batch.getWarehouseId())
|| !Objects.equals(lock.getPalletId(), item.getPalletId())
|| !Objects.equals(lock.getLocationId(), item.getLocationId())
|| !BUSINESS_TYPE.equals(lock.getBusinessType())
|| !item.getBusinessKey().equals(lock.getBusinessId())) {
throw conflict("Inbound plan pallet reservation is no longer active");
}
WarehousePalletBindingEntity binding = repository.lockPalletBinding(item.getCustomerId(),
batch.getWarehouseId(), BUSINESS_TYPE, item.getBusinessKey());
if (binding == null) {
binding = new WarehousePalletBindingEntity();
binding.setId(IdWorker.getId());
binding.setCustomerId(item.getCustomerId());
binding.setWarehouseId(batch.getWarehouseId());
binding.setBusinessType(BUSINESS_TYPE);
binding.setBusinessId(item.getBusinessKey());
binding.setPalletId(item.getPalletId());
binding.setCreatedAt(LocalDateTime.now());
repository.insertPalletBinding(binding);
} else if (!Objects.equals(binding.getPalletId(), item.getPalletId())) {
throw conflict("Inbound plan pallet business key is already bound to another pallet");
}
item.setStatus("CONFIRMED");
item.setUpdatedAt(LocalDateTime.now());
items.updateById(item);
}
batch.setStatus("CONFIRMED");
batch.setConfirmCommandKey(normalized.commandKey());
batch.setConfirmCommandHash(commandHash);
batch.setConfirmedAt(LocalDateTime.now());
batch.setUpdatedAt(batch.getConfirmedAt());
batches.updateById(batch);
return response(batch, batchItems, false);
}
@Transactional
public WarehousePalletBatchResponse compensate(WarehousePalletBatchRequests.Compensate request) {
WarehousePalletBatchRequests.Compensate normalized = normalize(request);
String commandHash = hash(normalized);
WarehousePalletBatchEntity batch = requiredBatch(normalized.warehouseId(), normalized.batchKey());
List<WarehousePalletBatchItemEntity> batchItems = items.lockByBatch(batch.getId());
if ("COMPENSATED".equals(batch.getStatus())) {
requireSameCommand(batch.getCompensateCommandHash(), commandHash);
return response(batch, batchItems, true);
}
if (!"RESERVED".equals(batch.getStatus())) {
throw conflict("Confirmed or subsequently occupied pallets cannot be compensated");
}
for (WarehousePalletBatchItemEntity item : batchItems) {
WarehousePalletBindingEntity binding = repository.lockPalletBinding(item.getCustomerId(),
batch.getWarehouseId(), BUSINESS_TYPE, item.getBusinessKey());
if (binding != null && Objects.equals(binding.getPalletId(), item.getPalletId())) {
throw conflict("A bound inbound plan pallet cannot be compensated");
}
locks.release(item.getCustomerId(), item.getResourceLockId());
item.setStatus("RELEASED");
item.setUpdatedAt(LocalDateTime.now());
items.updateById(item);
}
batch.setStatus("COMPENSATED");
batch.setCompensateCommandKey(normalized.commandKey());
batch.setCompensateCommandHash(commandHash);
batch.setCompensateReason(normalized.reason());
batch.setCompensatedAt(LocalDateTime.now());
batch.setUpdatedAt(batch.getCompensatedAt());
batches.updateById(batch);
return response(batch, batchItems, false);
}
private WarehousePalletBatchResponse reactivate(WarehousePalletBatchEntity batch) {
List<WarehousePalletBatchItemEntity> batchItems = items.lockByBatch(batch.getId());
int attempt = number(batch.getAttemptNo()) + 1;
batch.setAttemptNo(attempt);
for (WarehousePalletBatchItemEntity item : batchItems) {
if (repository.lockPalletBinding(item.getCustomerId(), batch.getWarehouseId(), BUSINESS_TYPE,
item.getBusinessKey()) != null) {
throw conflict("Inbound plan pallet business key became bound after compensation");
}
}
for (WarehousePalletBatchItemEntity item : batchItems) {
WarehousePalletEntity pallet = resources.pallet(WarehouseResourceService.PUBLIC_RESOURCE_SCOPE,
batch.getWarehouseId(), item.getPalletId());
WarehouseResourceLockResponse lock = locks.reserve(new WarehouseResourceLockRequests.ReserveRequest(
item.getCustomerId(), batch.getWarehouseId(), pallet.getId(), item.getLocationId(), BUSINESS_TYPE,
item.getBusinessKey(), reservationKey(batch.getId(), attempt, item.getSequenceNo())));
item.setResourceLockId(lock.lockId());
item.setStatus("RESERVED");
item.setUpdatedAt(LocalDateTime.now());
items.updateById(item);
}
batch.setStatus("RESERVED");
batch.setCompensateCommandKey(null);
batch.setCompensateCommandHash(null);
batch.setCompensateReason(null);
batch.setCompensatedAt(null);
batch.setUpdatedAt(LocalDateTime.now());
batches.updateById(batch);
return response(batch, batchItems, false);
}
private WarehousePalletBatchEntity requiredBatch(long warehouseId, String batchKey) {
WarehousePalletBatchEntity batch = batches.lockByKey(warehouseId, required(batchKey, "Batch key is required"));
if (batch == null) throw new BizException(ResultCode.NOT_FOUND, "Warehouse pallet batch not found");
return batch;
}
private static WarehousePalletBatchRequests.EnsureReserved normalize(
WarehousePalletBatchRequests.EnsureReserved request) {
if (request == null || request.warehouseId() <= 0) {
throw badRequest("Warehouse is required");
}
String batchKey = required(request.batchKey(), "Batch key is required");
requireMax(batchKey, 160, "Batch key is too long");
if (request.items() == null || request.items().isEmpty() || request.items().size() > 200) {
throw badRequest("One to 200 pallet items are required");
}
Set<String> businessKeys = new HashSet<>();
Set<Integer> sequences = new HashSet<>();
List<WarehousePalletBatchRequests.Item> normalized = request.items().stream().map(item -> {
if (item == null || item.customerId() <= 0 || item.sequenceNo() <= 0) {
throw badRequest("Pallet customer and sequence must be positive");
}
String businessKey = required(item.businessKey(), "Pallet business key is required");
String inboundPlanNo = required(item.inboundPlanNo(), "Inbound plan number is required");
String palletType = required(item.palletType(), "Pallet type is required");
requireMax(businessKey, 100, "Pallet business key is too long");
requireMax(inboundPlanNo, 128, "Inbound plan number is too long");
requireMax(palletType, 32, "Pallet type is too long");
if (item.fbaWarehouseCode() != null) {
requireMax(item.fbaWarehouseCode().trim(), 128, "FBA warehouse code is too long");
}
if (!businessKeys.add(businessKey)) throw badRequest("Pallet business key must be unique");
if (!sequences.add(item.sequenceNo())) throw badRequest("Pallet sequence must be unique");
return new WarehousePalletBatchRequests.Item(item.customerId(), businessKey, inboundPlanNo,
item.sequenceNo(), palletType,
trim(item.fbaWarehouseCode()));
}).sorted(Comparator.comparingInt(WarehousePalletBatchRequests.Item::sequenceNo)).toList();
return new WarehousePalletBatchRequests.EnsureReserved(request.warehouseId(), batchKey, normalized);
}
private static WarehousePalletBatchRequests.Confirm normalize(WarehousePalletBatchRequests.Confirm request) {
if (request == null || request.warehouseId() <= 0) throw badRequest("Warehouse is required");
String batchKey = required(request.batchKey(), "Batch key is required");
String commandKey = required(request.commandKey(), "Confirm command key is required");
requireMax(batchKey, 160, "Batch key is too long");
requireMax(commandKey, 160, "Confirm command key is too long");
return new WarehousePalletBatchRequests.Confirm(request.warehouseId(), batchKey, commandKey);
}
private static WarehousePalletBatchRequests.Compensate normalize(
WarehousePalletBatchRequests.Compensate request) {
if (request == null || request.warehouseId() <= 0) throw badRequest("Warehouse is required");
String batchKey = required(request.batchKey(), "Batch key is required");
String commandKey = required(request.commandKey(), "Compensation command key is required");
String reason = required(request.reason(), "Compensation reason is required");
requireMax(batchKey, 160, "Batch key is too long");
requireMax(commandKey, 160, "Compensation command key is too long");
requireMax(reason, 500, "Compensation reason is too long");
return new WarehousePalletBatchRequests.Compensate(request.warehouseId(), batchKey, commandKey, reason);
}
private static void requireSameRequest(WarehousePalletBatchEntity batch, long warehouseId, String requestHash) {
if (batch.getWarehouseId() != warehouseId || !requestHash.equals(batch.getRequestHash())) {
throw conflict("Batch key was already used with a different pallet request");
}
}
private static void requireSameCommand(String storedHash, String requestHash) {
if (!Objects.equals(storedHash, requestHash)) {
throw conflict("Terminal pallet batch command cannot be replayed with a different payload");
}
}
private WarehousePalletBatchResponse response(WarehousePalletBatchEntity batch,
List<WarehousePalletBatchItemEntity> batchItems,
boolean replayed) {
return new WarehousePalletBatchResponse(batch.getId(), batch.getBatchKey(), batch.getStatus(),
number(batch.getAttemptNo()), replayed, batchItems.stream().map(item ->
new WarehousePalletBatchResponse.Item(item.getCustomerId(), item.getBusinessKey(), item.getInboundPlanNo(),
item.getSequenceNo(), item.getPalletId(), item.getPalletCode(), item.getLocationId(),
item.getResourceLockId(), item.getStatus())).toList());
}
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("Cannot serialize warehouse pallet batch request", exception);
}
}
private static String reservationKey(WarehousePalletBatchEntity batch, int sequenceNo) {
return reservationKey(batch.getId(), number(batch.getAttemptNo()), sequenceNo);
}
private static String reservationKey(long batchId, int attemptNo, int sequenceNo) {
return "INBOUND_PLAN_AUTO:" + batchId + ":" + attemptNo + ":" + sequenceNo;
}
private static int number(Integer value) { return value == null ? 0 : value; }
private static String trim(String value) { return value == null || value.isBlank() ? null : value.trim(); }
private static String required(String value, String message) {
String normalized = trim(value);
if (normalized == null) throw badRequest(message);
return normalized;
}
private static void requireMax(String value, int max, String message) {
if (value.length() > max) throw badRequest(message);
}
private static BizException conflict(String message) { return new BizException(ResultCode.CONFLICT, message); }
private static BizException badRequest(String message) { return new BizException(ResultCode.BAD_REQUEST, message); }
}