This commit is contained in:
2026-10-20 14:22:31 +08:00
parent 19745e46c1
commit a870b920fe
95 changed files with 2616 additions and 275 deletions
@@ -0,0 +1,58 @@
package com.baiyue.lastleg.palletops.service;
import com.baiyue.lastleg.palletops.repository.PalletOperationRecoveryRepository;
import com.baiyue.lastleg.palletops.entity.PdaPalletOperation;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.server.ResponseStatusException;
import org.springframework.http.HttpStatus;
import jakarta.validation.constraints.*;
import java.time.LocalDateTime;
/** 只关闭已超时且从未发出远端请求的换标组托,不把未知远端结果推定为无效果。 */
@Service
public class PalletOperationRecoveryService {
private final PalletOperationRecoveryRepository repository;
private final ObjectMapper json;
/** 注入本域作业仓储和统一序列化器。 */
public PalletOperationRecoveryService(PalletOperationRecoveryRepository repository,ObjectMapper json){this.repository=repository;this.json=json;}
public record Close(@Positive long warehouseId,@NotNull LocalDateTime expectedUpdatedAt,
@NotBlank @Size(max=255) String idempotencyKey,@NotBlank @Size(max=1000) String reason){}
public record Recovery(String operationId,String status,String recoveryKey,String operatorId,String reason,
String previousStatus,String previousError,LocalDateTime closedAt,boolean replayed){}
public record State(String operationId,String status,String operatorKey,long warehouseId,String operationType,
String requestHash,String remoteTransitionsJson,String lastError,LocalDateTime updatedAt){}
/** 管理员只读核对原记录和不可变请求哈希,不返回其他账号会话。 */
@Transactional(readOnly=true)
public State inspect(long id){var row=required(repository.find(id));return new State(String.valueOf(id),row.getStatus(),row.getOperatorKey(),row.getOverseasWarehouseId(),row.getOperationType(),row.getRequestHash(),row.getRemoteTransitionsJson(),row.getLastError(),row.getUpdatedAt());}
/** 行锁与远端意图CAS互斥;审计写在原作业结果中,原错误和请求哈希保留。 */
@Transactional
public Recovery close(long id,String operatorId,Close request){
var row=required(repository.lock(id));
if(row.getOverseasWarehouseId()!=request.warehouseId())throw conflict("作业不属于指定仓库");
if("RECOVERED_NO_EFFECT".equals(row.getStatus())){
var existing=read(row.getResultJson());
if(!existing.recoveryKey().equals(request.idempotencyKey())||!existing.reason().equals(request.reason())||!existing.operatorId().equals(operatorId))throw conflict("恢复幂等键或载荷不一致");
return new Recovery(existing.operationId(),existing.status(),existing.recoveryKey(),existing.operatorId(),existing.reason(),existing.previousStatus(),existing.previousError(),existing.closedAt(),true);
}
// 已补偿失败仍须核对没有任何远端意图,不能仅凭补偿状态推断无效果。
if(!"RELABEL_PALLET_DISASSEMBLE".equals(row.getOperationType())||!("PROCESSING".equals(row.getStatus())||"FAILED_COMPENSATED".equals(row.getStatus())))throw conflict("只允许恢复超时的处理中或已补偿失败换标组托");
if(!request.expectedUpdatedAt().equals(row.getUpdatedAt())||row.getUpdatedAt()==null||row.getUpdatedAt().isAfter(LocalDateTime.now().minusMinutes(15)))throw conflict("作业仍活跃或版本已变化,请刷新后核对");
if(row.getResultJson()!=null&&!row.getResultJson().isBlank())throw conflict("作业已有结果,不能关闭为无效果");
try{if(row.getRemoteTransitionsJson()!=null&&!row.getRemoteTransitionsJson().isBlank()){
var attempts=json.readTree(row.getRemoteTransitionsJson());
if(!attempts.isArray()||!attempts.isEmpty())throw conflict("存在远端意图或回执,必须先核实远端效果,不能执行无效果恢复");
}}catch(com.fasterxml.jackson.core.JsonProcessingException e){throw conflict("远端意图记录不可解析,禁止恢复");}
var result=new Recovery(String.valueOf(id),"RECOVERED_NO_EFFECT",request.idempotencyKey(),operatorId,request.reason(),row.getStatus(),row.getLastError(),LocalDateTime.now(),false);
try{if(!repository.close(id,json.writeValueAsString(result),result.closedAt()))throw conflict("作业状态已变化,恢复未写入");}
catch(com.fasterxml.jackson.core.JsonProcessingException e){throw new IllegalStateException("恢复审计序列化失败",e);}
return result;
}
/** 重放只读取已持久化的恢复审计。 */
private Recovery read(String value){try{return json.readValue(value,Recovery.class);}catch(Exception e){throw conflict("恢复审计不可读取");}}
/** 原记录不存在时明确拒绝,不新建作业。 */
private PdaPalletOperation required(PdaPalletOperation row){if(row==null)throw new ResponseStatusException(HttpStatus.NOT_FOUND,"托盘作业不存在");return row;}
/** 统一表达恢复条件或幂等载荷冲突。 */
private ResponseStatusException conflict(String message){return new ResponseStatusException(HttpStatus.CONFLICT,message);}
}
@@ -54,6 +54,7 @@ import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@@ -63,6 +64,7 @@ import java.util.stream.Collectors;
public class PdaPalletOperationService {
@Resource private PdaPalletOperationMapper mapper;
@Resource private PdaPalletOperationRepository mutationRepository;
@Resource private com.baiyue.lastleg.palletops.repository.PalletOperationRecoveryRepository recoveryRepository;
@Resource private PhysicalPalletSerialPort physicalPalletSerial;
private Clock mutationClock = Clock.systemUTC();
@Resource private com.baiyue.lastleg.inventory.command.mapper.InventoryCommandMapper inventoryCommands;
@@ -257,7 +259,10 @@ public class PdaPalletOperationService {
List<RelabelItem> selected = relabelItems.selectList(new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<RelabelItem>()
.eq("application_id", application.getId()).eq("customer_id", application.getCustomerId())
.in("carton_mark", marks));
if (selected.size() != marks.size()
// 同一退货实物来源允许多个SKU按件任务;普通箱任务仍必须一箱一项。
boolean bulkSkuSource = selected.stream().allMatch(item -> item.getReturnCartonId() != null
&& "RETURN_BULK_SKU_LABEL".equals(item.getOperationType()));
if ((!bulkSkuSource && selected.size() != marks.size())
|| selected.stream().map(RelabelItem::getCartonMark).distinct().count() != marks.size()) {
throw conflict("提交的箱唛不属于当前换标单");
}
@@ -417,8 +422,10 @@ public class PdaPalletOperationService {
// 先取得共同串行锁再触碰远端;本地失败补偿也必须在该锁释放前结束。
try {
applyPhysical(operation,identity.overseasWarehouseId(),target,"OCCUPIED",target.locationId(),key,"退货换标组托",journal);
Set<Long> movedCartons = new HashSet<>();
for (int index = 0; index < cartons.size(); index++) {
RelabelItem item = selected.get(index);
if (!movedCartons.add(item.getReturnCartonId())) continue;
ReturnInventoryCarton carton = returnCartons.lockId(item.getReturnCartonId());
if (carton == null || !Objects.equals(carton.getCustomerId(), application.getCustomerId())
|| (unboundSource ? (org.springframework.util.StringUtils.hasText(carton.getPalletCode()) && !"UNBOUND".equalsIgnoreCase(carton.getPalletCode()))
@@ -426,8 +433,18 @@ public class PdaPalletOperationService {
|| !Objects.equals(carton.getRelabelApplicationId(), application.getId())
|| !"RELABEL_PENDING".equals(carton.getStatus()))
throw conflict("退货换标来源库存已变化,请刷新后重试");
var sameSource = selected.stream().filter(line -> Objects.equals(line.getReturnCartonId(), carton.getId())).toList();
long move = "RETURN_BULK_SKU_LABEL".equals(item.getOperationType())
? value(item.getNotifiedPackageCount()) : value(carton.getTotalQuantity());
? sameSource.stream().mapToLong(line -> value(line.getNotifiedPackageCount())).sum() : value(carton.getTotalQuantity());
if (sameSource.size() > 1) {
// 同一实物的多SKU整批只移动一次;部分混源必须先拆成独立实物,不能逐SKU虚构新箱。
var requestedSkus = new HashMap<String, Long>();
sameSource.forEach(line -> requestedSkus.merge(line.getSkuCode().toUpperCase(Locale.ROOT), (long) value(line.getNotifiedPackageCount()), Long::sum));
var actualSkus = new HashMap<String, Long>();
returnCartonItems.lockCarton(carton.getId()).forEach(line -> actualSkus.merge(line.getSkuCode().toUpperCase(Locale.ROOT), value(line.getQuantity()), Long::sum));
if (move != value(carton.getTotalQuantity()) || !requestedSkus.equals(actualSkus))
throw conflict("多SKU来源须整批完成并核平;部分贴标请先分拣为独立实物");
}
if (move <= 0 || move > value(carton.getTotalQuantity())) throw conflict("来源托盘可拆数量不足");
if (move < value(carton.getTotalQuantity())) {
splitReturnRelabelCarton(application, item, carton, target, move);
@@ -439,7 +456,7 @@ public class PdaPalletOperationService {
returnCartons.updateById(carton);
}
}
recordReturnRelabelBinding(identity, application, selected, target.palletCode());
recordReturnRelabelBinding(identity, application, selected, source == null ? "" : source.palletCode(), target.palletCode());
List<PalletOperationResponses.PalletContent> contents = source == null
? List.of(content(identity.overseasWarehouseId(), physical.resolve(target.palletCode())))
: List.of(content(identity.overseasWarehouseId(), physical.resolve(source.palletCode())),
@@ -972,6 +989,7 @@ public class PdaPalletOperationService {
return marks;
}
/** 记录经过实物来源校验的逐任务绑定,避免多来源数组失去箱与托盘的精确关系。 */
private void recordRelabelBinding(PdaWarehouseIdentity identity, RelabelMutation mutation,
Map<Long, InventoryPosition> targets) {
var app = mutation.application();
@@ -987,7 +1005,11 @@ public class PdaPalletOperationService {
mutation.itemsByPosition().values().stream().flatMap(List::stream).toList(),
mutation.itemsByPosition().keySet().stream().map(positions::selectById).filter(Objects::nonNull).findFirst().orElseThrow().getPalletId(),
targets.values().iterator().next().getPalletId(), targets.values().iterator().next().getPalletNo(), mutation.measuredCartons());
record.setDetailJson(write(Map.of("round", Objects.toString(task.getReviewedAt(), "INITIAL"),
var itemBindings = mutation.itemsByPosition().entrySet().stream().flatMap(entry -> entry.getValue().stream().map(item ->
Map.of("itemId", String.valueOf(item.getId()), "sourceKnown", true,
"sourcePalletCode", Objects.toString(positions.selectById(entry.getKey()).getPalletNo(), ""),
"targetPalletCode", targets.get(entry.getKey()).getPalletNo()))).toList();
record.setDetailJson(write(Map.of("round", Objects.toString(task.getReviewedAt(), "INITIAL"), "itemBindings", itemBindings,
"measuredCartons", measured,
"cartonMarks", mutation.itemsByPosition().values().stream().flatMap(List::stream).map(RelabelItem::getCartonMark).toList(),
"targetPallets", targets.values().stream().map(InventoryPosition::getPalletNo).distinct().toList(),
@@ -996,8 +1018,9 @@ public class PdaPalletOperationService {
relabelWorkHistory.insert(record);
}
/** 来源实托在变更前已经核对,历史只记录该次校验值,不反查变更后的库存托盘。 */
private void recordReturnRelabelBinding(PdaWarehouseIdentity identity, RelabelApplication app,
List<RelabelItem> items, String targetPallet) {
List<RelabelItem> items, String sourcePallet, String targetPallet) {
var task = relabelWorkTasks.find(app.getCustomerId(), app.getId());
if (task == null) throw conflict("退货换标作业任务不存在,不能记录组托");
var record = new com.baiyue.lastleg.relabel.pda.entity.PdaRelabelHistory();
@@ -1007,6 +1030,8 @@ public class PdaPalletOperationService {
record.setOperatorKey(identity.operatorKey()); record.setOperatorName(identity.operatorName());
record.setCreatedAt(LocalDateTime.now());
record.setDetailJson(write(Map.of("round", Objects.toString(task.getReviewedAt(), "INITIAL"),
"itemBindings", items.stream().map(item -> Map.of("itemId", String.valueOf(item.getId()), "sourceKnown", true,
"sourcePalletCode", sourcePallet, "targetPalletCode", targetPallet)).toList(),
"cartonMarks", items.stream().map(RelabelItem::getCartonMark).toList(),
"targetPallets", List.of(targetPallet))));
relabelWorkHistory.insert(record);
@@ -1200,17 +1225,18 @@ public class PdaPalletOperationService {
newTransaction.executeWithoutResult(status -> {
PdaPalletOperation operation = mapper.selectById(operationId);
if (operation == null) throw new IllegalStateException("实托作业意图记录不存在,不能发出远端请求");
operation.setRemoteTransitionsJson(write(attempts)); operation.setUpdatedAt(LocalDateTime.now());
mapper.updateById(operation);
if (!recoveryRepository.journal(operationId, write(attempts), LocalDateTime.now()))
throw conflict("作业已由管理员恢复关闭,禁止继续发出远端请求");
});
}
/** 条件记录失败,避免旧线程覆盖管理员已恢复的原记录。 */
private void fail(long operationId, RuntimeException failure, boolean compensated) {
newTransaction.executeWithoutResult(status -> {
PdaPalletOperation operation = mapper.selectById(operationId);
if (operation == null || "COMPLETED".equals(operation.getStatus())) return;
if (operation == null || "COMPLETED".equals(operation.getStatus()) || "RECOVERED_NO_EFFECT".equals(operation.getStatus())) return;
operation.setStatus(compensated ? "FAILED_COMPENSATED" : "COMPENSATION_REQUIRED");
operation.setLastError(limit(failure.getMessage(), 2000)); operation.setUpdatedAt(LocalDateTime.now());
mapper.updateById(operation);
recoveryRepository.fail(operationId, operation.getStatus(), operation.getLastError(), operation.getUpdatedAt());
});
}
private boolean compensate(long warehouseId, List<PalletPhysicalGateway.Transition> transitions) {