feat(lastleg): redispatch returned cargo and reconcile source signatures
This commit is contained in:
+59
@@ -0,0 +1,59 @@
|
||||
package com.baiyue.lastleg.delivery.mapper;
|
||||
|
||||
import com.baiyue.lastleg.delivery.model.SourceDeliveryCompletionPolicy.*;
|
||||
import org.apache.ibatis.annotations.*;
|
||||
import java.util.List;
|
||||
|
||||
/** Use SourceDeliveryCompletionQuery in an explicit READ_COMMITTED write transaction.
|
||||
* lockSource is a current locking read; history queries deliberately do not lock delivery rows.
|
||||
* No pagination or current ll_order.delivery_order_no association is used.
|
||||
*/
|
||||
@Mapper
|
||||
public interface DeliverySourceEvidenceMapper {
|
||||
@Select("SELECT source_inbound_id, id AS order_id, package_count AS packages FROM ll_order " +
|
||||
"WHERE customer_id=#{customerId} AND source_region='FIRSTLEG' AND source_inbound_id=#{sourceInboundId}")
|
||||
Source source(@Param("customerId") String customerId, @Param("sourceInboundId") String sourceInboundId);
|
||||
|
||||
@Select("SELECT source_inbound_id,id AS order_id,package_count AS packages FROM ll_order " +
|
||||
"WHERE customer_id=#{customerId} AND source_region='FIRSTLEG' AND id=#{orderId} FOR UPDATE")
|
||||
Source lockSource(@Param("customerId") String customerId, @Param("orderId") long orderId);
|
||||
|
||||
@Select("SELECT DISTINCT t.dispatch_order_id FROM ll_dispatch_pda_pick_order a " +
|
||||
"JOIN ll_dispatch_pda_pick p ON p.customer_id=a.customer_id AND p.id=a.pick_id " +
|
||||
"JOIN ll_dispatch_pda_task t ON t.customer_id=p.customer_id AND t.id=p.task_id " +
|
||||
"WHERE a.customer_id=#{customerId} AND a.order_id=#{orderId} ORDER BY t.dispatch_order_id")
|
||||
List<Long> dispatchIds(@Param("customerId") String customerId, @Param("orderId") long orderId);
|
||||
|
||||
@Select("SELECT d.id AS delivery_id,d.dispatch_order_id AS dispatch_id,d.delivery_group_key AS group_key," +
|
||||
"d.package_count AS packages,d.signed_packages,d.signed_at,ds.depart_at AS departed_at FROM ll_delivery_order d " +
|
||||
"LEFT JOIN ll_dispatch_order ds ON ds.customer_id=d.customer_id AND ds.id=d.dispatch_order_id " +
|
||||
"WHERE d.customer_id=#{customerId} AND d.dispatch_order_id=#{dispatchId} ORDER BY d.id")
|
||||
List<Delivery> deliveries(@Param("customerId") String customerId, @Param("dispatchId") long dispatchId);
|
||||
|
||||
@Select("SELECT a.id AS attribution_id,t.dispatch_order_id AS dispatch_id," +
|
||||
"CASE WHEN di.destination_type='PRIVATE' AND NULLIF(TRIM(di.delivery_address),'') IS NOT NULL " +
|
||||
"THEN CONCAT('PRIVATE:',TRIM(di.delivery_address),':',TRIM(COALESCE(di.postal_code,''))) " +
|
||||
"WHEN COALESCE(NULLIF(TRIM(di.destination_warehouse_code),''),NULLIF(TRIM(di.fba_warehouse_code),'')) IS NOT NULL " +
|
||||
"THEN CONCAT('CODE:',UPPER(COALESCE(NULLIF(TRIM(di.destination_warehouse_code),''),NULLIF(TRIM(di.fba_warehouse_code),'')))) " +
|
||||
"WHEN NULLIF(TRIM(di.destination_warehouse_id),'') IS NOT NULL THEN CONCAT('ID:',TRIM(di.destination_warehouse_id)) ELSE NULL END AS group_key," +
|
||||
"a.order_id,a.packages,a.receipt_id,a.return_receipt_id,r.inventory_ledger_id AS return_ledger_id,r.received_packages AS return_packages FROM ll_dispatch_pda_pick_order a " +
|
||||
"JOIN ll_dispatch_pda_pick p ON p.customer_id=a.customer_id AND p.id=a.pick_id " +
|
||||
"JOIN ll_dispatch_pda_task t ON t.customer_id=p.customer_id AND t.id=p.task_id " +
|
||||
"LEFT JOIN ll_dispatch_item di ON di.customer_id=p.customer_id AND di.id=p.dispatch_item_id AND di.dispatch_order_id=t.dispatch_order_id " +
|
||||
"LEFT JOIN ll_return_stock_receipt r ON r.customer_id=a.customer_id AND r.id=a.return_receipt_id AND r.order_id=a.order_id " +
|
||||
"AND EXISTS (SELECT 1 FROM ll_inventory_ledger rl JOIN ll_exception_order re ON re.customer_id=rl.customer_id " +
|
||||
"AND re.id=r.exception_order_id AND re.exception_no=rl.business_order_no AND re.last_mile_order_id=a.order_id " +
|
||||
"WHERE rl.customer_id=r.customer_id AND rl.id=r.inventory_ledger_id AND rl.business_type='RETURN_STOCK_IN' " +
|
||||
"AND rl.outbound_order_no=r.source_outbound_no AND rl.delta_packages=r.received_packages " +
|
||||
"AND rl.source_event_id=CONCAT('RETURN_SOURCE:',r.source_outbound_pallet_item_id,':',r.source_inventory_position_id)) " +
|
||||
"WHERE a.customer_id=#{customerId} AND t.dispatch_order_id=#{dispatchId} ORDER BY a.id")
|
||||
List<Pick> picks(@Param("customerId") String customerId, @Param("dispatchId") long dispatchId);
|
||||
|
||||
@Select("SELECT l.id AS ledger_id,d.id AS delivery_id,e.last_mile_order_id AS order_id," +
|
||||
"l.delta_packages AS packages,l.occurred_at AS received_at FROM ll_delivery_order d " +
|
||||
"JOIN ll_exception_order e ON e.customer_id=d.customer_id AND e.delivery_order_id=d.id " +
|
||||
"JOIN ll_inventory_ledger l ON l.customer_id=e.customer_id AND l.business_order_no=e.exception_no " +
|
||||
"AND l.outbound_order_no=d.outbound_no AND l.business_type='RETURN_STOCK_IN' AND l.delta_packages>0 " +
|
||||
"WHERE d.customer_id=#{customerId} AND d.dispatch_order_id=#{dispatchId} ORDER BY l.id")
|
||||
List<Returned> returns(@Param("customerId") String customerId, @Param("dispatchId") long dispatchId);
|
||||
}
|
||||
+134
@@ -0,0 +1,134 @@
|
||||
package com.baiyue.lastleg.delivery.model;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.*;
|
||||
|
||||
/** Read-only proof of whole-source signature. Inputs must cover ALL history for the requested source. */
|
||||
public final class SourceDeliveryCompletionPolicy {
|
||||
private SourceDeliveryCompletionPolicy() { }
|
||||
public record Source(String sourceInboundId, long orderId, long packages) { }
|
||||
public record Delivery(long deliveryId, long dispatchId, String groupKey, long packages,
|
||||
Long signedPackages, LocalDateTime signedAt, LocalDateTime departedAt) { }
|
||||
public record Pick(long attributionId, long dispatchId, String groupKey, long orderId, long packages,
|
||||
Long receiptId, Long returnReceiptId, Long returnLedgerId, Long returnPackages) { }
|
||||
public record Returned(long ledgerId, long deliveryId, long orderId, long packages, LocalDateTime receivedAt) { }
|
||||
public record Result(String sourceInboundId, long originalPackages, long provenSignedPackages,
|
||||
boolean fullySigned, List<String> unresolvedReasons) { }
|
||||
|
||||
public static Result evaluate(Source source, List<Delivery> deliveries, List<Pick> picks,
|
||||
List<Returned> returns, boolean completeHistory) {
|
||||
Objects.requireNonNull(source);
|
||||
var reasons = new LinkedHashSet<String>();
|
||||
if (!completeHistory) reasons.add("HISTORY_NOT_COMPLETE");
|
||||
if (source.sourceInboundId() == null || source.sourceInboundId().isBlank() || source.packages() <= 0)
|
||||
reasons.add("SOURCE_QUANTITY_MISSING");
|
||||
// Only destinations containing this source can affect its proof. Keep co-shipped sources
|
||||
// within those destinations to reconcile the exact delivery signature total.
|
||||
var relevant = deliveries.stream().filter(d -> picks.stream().anyMatch(p -> p.orderId() == source.orderId()
|
||||
&& p.dispatchId() == d.dispatchId() && Objects.equals(p.groupKey(),d.groupKey()))).toList();
|
||||
var relevantIds = new HashSet<Long>();
|
||||
for (var d : relevant) if (!relevantIds.add(d.deliveryId())) reasons.add("DUPLICATE_DELIVERY");
|
||||
var attributionIds = new HashSet<Long>(); var ledgerIds = new HashSet<Long>();
|
||||
for (var p : picks) if (relevant.stream().anyMatch(d -> d.dispatchId()==p.dispatchId() && Objects.equals(d.groupKey(),p.groupKey()))) {
|
||||
if (!attributionIds.add(p.attributionId()) || p.packages() <= 0 || p.groupKey() == null) reasons.add("INVALID_PICK_ATTRIBUTION");
|
||||
if ((p.receiptId() == null) == (p.returnReceiptId() == null)
|
||||
|| (p.receiptId()!=null && p.receiptId()<=0)) reasons.add("PICK_RECEIPT_ORIGIN_UNRESOLVED");
|
||||
}
|
||||
for (var r : returns) if (relevantIds.contains(r.deliveryId()) && (!ledgerIds.add(r.ledgerId())
|
||||
|| r.packages()<=0 || r.receivedAt()==null)) reasons.add("INVALID_RETURN_LEDGER");
|
||||
long signed = 0;
|
||||
for (var d : relevant) {
|
||||
var matching = picks.stream().filter(p -> p.dispatchId() == d.dispatchId() && Objects.equals(p.groupKey(), d.groupKey())).toList();
|
||||
var quantities = new HashMap<Long, Long>();
|
||||
for (var p : matching) quantities.merge(p.orderId(), p.packages(), Math::addExact);
|
||||
if (d.groupKey() == null || d.packages() <= 0 || d.departedAt() == null
|
||||
|| matching.stream().mapToLong(Pick::packages).sum() != d.packages()
|
||||
|| deliveries.stream().filter(other -> other.dispatchId() == d.dispatchId() && Objects.equals(other.groupKey(), d.groupKey())).count() != 1) {
|
||||
reasons.add("DELIVERY_SOURCE_COVERAGE_UNRESOLVED:" + d.deliveryId()); continue;
|
||||
}
|
||||
if (d.signedAt() == null) {
|
||||
if (d.signedPackages() != null) reasons.add("SIGNATURE_TIME_MISSING:" + d.deliveryId());
|
||||
continue;
|
||||
}
|
||||
if (d.signedAt().isBefore(d.departedAt()) || d.signedPackages() == null || d.signedPackages() <= 0 || d.signedPackages() > d.packages()) {
|
||||
reasons.add("SIGNATURE_QUANTITY_UNRESOLVED:" + d.deliveryId()); continue;
|
||||
}
|
||||
var before = new HashMap<Long, Long>(); boolean valid = true;
|
||||
for (var r : returns) if (r.deliveryId() == d.deliveryId()) {
|
||||
if (r.receivedAt() == null || r.receivedAt().isBefore(d.departedAt()) || !quantities.containsKey(r.orderId())) {
|
||||
valid = false; break;
|
||||
}
|
||||
// Millisecond-equal timestamps cannot prove which transaction happened first.
|
||||
if (r.receivedAt().equals(d.signedAt())) { valid = false; break; }
|
||||
if (r.receivedAt().isBefore(d.signedAt())) before.merge(r.orderId(), r.packages(), Math::addExact);
|
||||
}
|
||||
if (!valid || before.entrySet().stream().anyMatch(e -> e.getValue() > quantities.getOrDefault(e.getKey(), 0L))
|
||||
|| d.packages() - before.values().stream().mapToLong(Long::longValue).sum() != d.signedPackages()) {
|
||||
reasons.add("PRE_SIGNATURE_RETURN_UNRESOLVED:" + d.deliveryId()); continue;
|
||||
}
|
||||
long fresh = 0, previouslySigned = 0; boolean lineageValid = true;
|
||||
for (var p : matching) if (p.orderId() == source.orderId()) {
|
||||
Boolean eligible = freshOrigin(p,d,deliveries,picks,returns,new HashSet<>(),reasons);
|
||||
if (eligible == null) lineageValid = false;
|
||||
else if (eligible) fresh = Math.addExact(fresh,p.packages());
|
||||
else previouslySigned = Math.addExact(previouslySigned,p.packages());
|
||||
}
|
||||
long returnedBefore = before.getOrDefault(source.orderId(),0L);
|
||||
if (fresh>0 && previouslySigned>0 && returnedBefore>0) {
|
||||
reasons.add("MIXED_RESIGNED_RETURN_UNRESOLVED:"+d.deliveryId()); lineageValid=false;
|
||||
}
|
||||
if (lineageValid) signed = Math.addExact(signed,fresh==0?0:fresh-returnedBefore);
|
||||
}
|
||||
for (var p : picks) if (p.orderId() == source.orderId() && deliveries.stream().noneMatch(d ->
|
||||
d.dispatchId() == p.dispatchId() && Objects.equals(d.groupKey(), p.groupKey()))) reasons.add("PICK_DELIVERY_MISSING");
|
||||
// Repeated post-signature shipments cannot be treated as distinct originally loaded boxes.
|
||||
if (signed > source.packages()) reasons.add("SIGNED_EXCEEDS_ORIGINAL_SOURCE");
|
||||
return new Result(source.sourceInboundId(), source.packages(), signed,
|
||||
reasons.isEmpty() && signed == source.packages(), List.copyOf(reasons));
|
||||
}
|
||||
/** Receipt identity is exact; package identity inside mixed fresh/already-signed cohorts is not. */
|
||||
private static Boolean freshOrigin(Pick pick, Delivery current, List<Delivery> deliveries, List<Pick> picks,
|
||||
List<Returned> returns, Set<Long> visiting, Set<String> reasons) {
|
||||
if (!visiting.add(pick.attributionId())) { reasons.add("RETURN_LINEAGE_CYCLE"); return null; }
|
||||
try {
|
||||
if ((pick.receiptId()==null)==(pick.returnReceiptId()==null)) { reasons.add("PICK_RECEIPT_ORIGIN_UNRESOLVED"); return null; }
|
||||
if (pick.receiptId()!=null) {
|
||||
if(pick.receiptId()<=0) { reasons.add("PICK_RECEIPT_ORIGIN_UNRESOLVED"); return null; }
|
||||
return true;
|
||||
}
|
||||
var receipts=returns.stream().filter(r->Objects.equals(pick.returnLedgerId(),r.ledgerId())).toList();
|
||||
if(pick.returnReceiptId()<=0 || receipts.size()!=1 || pick.returnPackages()==null
|
||||
|| receipts.get(0).orderId()!=pick.orderId() || receipts.get(0).packages()!=pick.returnPackages()
|
||||
|| picks.stream().filter(p->Objects.equals(p.returnLedgerId(),pick.returnLedgerId())).mapToLong(Pick::packages).sum()>pick.returnPackages()) {
|
||||
reasons.add("RETURN_PICK_ORIGIN_UNRESOLVED:"+pick.attributionId()); return null;
|
||||
}
|
||||
var receipt=receipts.get(0);
|
||||
var parents=deliveries.stream().filter(d->d.deliveryId()==receipt.deliveryId()).toList();
|
||||
if(parents.size()!=1 || receipt.receivedAt()==null || current.departedAt()==null
|
||||
|| !receipt.receivedAt().isBefore(current.departedAt())) {
|
||||
reasons.add("RETURN_PICK_CHRONOLOGY_UNRESOLVED:"+pick.attributionId());return null;
|
||||
}
|
||||
var parent=parents.get(0);
|
||||
if(parent.departedAt()==null || receipt.receivedAt().isBefore(parent.departedAt())
|
||||
|| (parent.signedAt()!=null && receipt.receivedAt().equals(parent.signedAt()))) {
|
||||
reasons.add("RETURN_PICK_LINEAGE_UNRESOLVED:"+pick.attributionId()); return null;
|
||||
}
|
||||
var parentPicks=picks.stream().filter(p->p.dispatchId()==parent.dispatchId()
|
||||
&& Objects.equals(p.groupKey(),parent.groupKey()) && p.orderId()==pick.orderId()).toList();
|
||||
if(parentPicks.isEmpty() || receipt.packages()>parentPicks.stream().mapToLong(Pick::packages).sum()) {
|
||||
reasons.add("RETURN_PICK_ORIGIN_UNRESOLVED:"+pick.attributionId());return null;
|
||||
}
|
||||
// After a validated signature every box remaining in that movement is already signed.
|
||||
if(parent.signedAt()!=null && parent.signedAt().isBefore(receipt.receivedAt())) return false;
|
||||
Boolean eligibility=null;
|
||||
for(var ancestor:parentPicks) {
|
||||
Boolean value=freshOrigin(ancestor,parent,deliveries,picks,returns,visiting,reasons);
|
||||
if(value==null)return null;
|
||||
if(eligibility!=null && !eligibility.equals(value)) { reasons.add("RETURN_LINEAGE_MIXED_COHORT");return null; }
|
||||
eligibility=value;
|
||||
}
|
||||
return eligibility;
|
||||
} finally { visiting.remove(pick.attributionId()); }
|
||||
}
|
||||
|
||||
}
|
||||
+17
-18
@@ -35,25 +35,14 @@ public class DeliveryCompletedOutboxService {
|
||||
|
||||
public enum Publication { PUBLISHED, NO_DOMESTIC_SOURCE, UNKNOWN_SIGNATURE_REQUIRES_RECONCILIATION, MIXED_RETURN_REQUIRES_SOURCE_RECONCILIATION }
|
||||
|
||||
public Publication publish(DeliveryOrder delivery, List<LastlegOrder> orders, Instant completedAt) {
|
||||
public Publication publish(DeliveryOrder delivery, List<LastlegOrder> orders, Map<String, com.baiyue.lastleg.delivery.model.SourceDeliveryCompletionPolicy.Result> proofs, Instant completedAt) {
|
||||
Objects.requireNonNull(delivery, "delivery");
|
||||
Objects.requireNonNull(completedAt, "completedAt");
|
||||
// The domestic consumer marks the whole inbound order DELIVERED (signed), not merely transport closed.
|
||||
// A mixed return/signature must not publish that full-signature fact.
|
||||
Map<String, LastlegOrder> bySourceInbound = new LinkedHashMap<>();
|
||||
for (LastlegOrder order : orders == null ? List.<LastlegOrder>of() : orders) {
|
||||
// Only orders projected from the domestic first-leg domain need an
|
||||
// acknowledgement event. Customer/local overseas orders can still carry
|
||||
// a local sourceInboundId, but deliberately have no domestic sourceOrderId.
|
||||
if (StringUtils.hasText(order.getSourceRegion())
|
||||
&& !"FIRSTLEG".equals(order.getSourceRegion())) continue;
|
||||
// Relabel and warehouse-transfer deliveries are created and completed wholly
|
||||
// in the overseas last-leg domain. They do not have a domestic source order
|
||||
// to acknowledge, so publishing the cross-region event would make receipt
|
||||
// completion fail on the missing source-order fields.
|
||||
if ("RELABEL_APPLICATION".equals(order.getOrderType())
|
||||
|| "WAREHOUSE_TRANSFER".equals(order.getOrderType())) continue;
|
||||
if (!StringUtils.hasText(order.getSourceInboundId())) continue;
|
||||
if (!domestic(order)) continue;
|
||||
requireText(order.getSourceOrderId(), "sourceOrderId");
|
||||
requireText(order.getInboundCode(), "inboundCode");
|
||||
requireText(order.getCustomerId(), "customerId");
|
||||
@@ -67,15 +56,25 @@ public class DeliveryCompletedOutboxService {
|
||||
if (delivery.getPackageCount() == null || delivery.getPackageCount() <= 0 || delivery.getSignedPackages() == null
|
||||
|| delivery.getSignedPackages() <= 0 || delivery.getSignedPackages() > delivery.getPackageCount())
|
||||
return Publication.UNKNOWN_SIGNATURE_REQUIRES_RECONCILIATION;
|
||||
if (!Objects.equals(delivery.getSignedPackages(), delivery.getPackageCount()))
|
||||
return Publication.MIXED_RETURN_REQUIRES_SOURCE_RECONCILIATION;
|
||||
for (LastlegOrder order : bySourceInbound.values()) publishOne(delivery, order, completedAt);
|
||||
return Publication.PUBLISHED;
|
||||
boolean pending = false;
|
||||
for (LastlegOrder order : bySourceInbound.values()) {
|
||||
var proof = proofs.get(order.getSourceInboundId());
|
||||
if (proof == null || !proof.fullySigned() || !proof.unresolvedReasons().isEmpty()
|
||||
|| !Objects.equals(proof.sourceInboundId(),order.getSourceInboundId())
|
||||
|| order.getPackageCount() == null || proof.originalPackages() != order.getPackageCount()
|
||||
|| proof.provenSignedPackages() != proof.originalPackages()) { pending = true; continue; }
|
||||
publishOne(delivery, order, completedAt);
|
||||
}
|
||||
return pending ? Publication.MIXED_RETURN_REQUIRES_SOURCE_RECONCILIATION : Publication.PUBLISHED;
|
||||
}
|
||||
static boolean domestic(LastlegOrder order) {
|
||||
return "FIRSTLEG".equals(order.getSourceRegion()) && StringUtils.hasText(order.getSourceInboundId())
|
||||
&& !"RELABEL_APPLICATION".equals(order.getOrderType()) && !"WAREHOUSE_TRANSFER".equals(order.getOrderType());
|
||||
}
|
||||
|
||||
private void publishOne(DeliveryOrder delivery, LastlegOrder order, Instant completedAt) {
|
||||
long aggregateVersion = Math.max(1L, delivery.getVersion() == null ? 1L : delivery.getVersion());
|
||||
String eventKey = EventTypes.OVERSEAS_DELIVERY_COMPLETED + ':' + delivery.getDeliveryNo()
|
||||
String eventKey = EventTypes.OVERSEAS_DELIVERY_COMPLETED + ':' + order.getCustomerId()
|
||||
+ ':' + order.getSourceInboundId();
|
||||
var payload = new OverseasDeliveryCompletedV1(String.valueOf(delivery.getId()), delivery.getDeliveryNo(),
|
||||
String.valueOf(order.getId()), order.getOrderNo(), order.getSourceInboundId(), order.getInboundCode(),
|
||||
|
||||
+52
-27
@@ -35,6 +35,8 @@ public class DeliveryDocumentService {
|
||||
@Resource private LastlegOrderMapper lastlegOrders;
|
||||
@Resource private DeliveryCompletedOutboxService deliveryCompletedOutbox;
|
||||
@Resource private CustomerVisitService customerVisits;
|
||||
@Resource private SourceDeliveryCompletionQuery sourceCompletion;
|
||||
@Resource private DeliverySourceEvidenceMapper sourceEvidence;
|
||||
|
||||
public Detail detail(String customerId, String deliveryNo) { return detail(require(customerId, deliveryNo), false); }
|
||||
|
||||
@@ -56,7 +58,7 @@ public class DeliveryDocumentService {
|
||||
}
|
||||
|
||||
/** Lock the same aggregate as status commands, validate the whole set, then commit one snapshot. */
|
||||
@Transactional
|
||||
@Transactional(isolation = org.springframework.transaction.annotation.Isolation.READ_COMMITTED)
|
||||
public Detail save(String customerId, String deliveryNo, String operatorId, Save request) {
|
||||
DeliveryOrder order = orders.lockByNo(customerId, deliveryNo);
|
||||
if (order == null) throw missing();
|
||||
@@ -110,19 +112,8 @@ public class DeliveryDocumentService {
|
||||
}
|
||||
orders.updateById(order);
|
||||
if (autoCompleted) {
|
||||
List<LastlegOrder> completedOrders = completeRelatedOrders(order, now);
|
||||
var publication = deliveryCompletedOutbox.publish(order, completedOrders, now.toInstant(ZoneOffset.UTC));
|
||||
if (publication == DeliveryCompletedOutboxService.Publication.MIXED_RETURN_REQUIRES_SOURCE_RECONCILIATION
|
||||
|| publication == DeliveryCompletedOutboxService.Publication.UNKNOWN_SIGNATURE_REQUIRES_RECONCILIATION) {
|
||||
DeliveryOperation pending = new DeliveryOperation();
|
||||
pending.setCustomerId(customerId); pending.setDeliveryOrderId(order.getId()); pending.setOperatorId(operatorId);
|
||||
pending.setOperationNode("FIRSTLEG_COMPLETION_PENDING");
|
||||
pending.setResult(publication == DeliveryCompletedOutboxService.Publication.UNKNOWN_SIGNATURE_REQUIRES_RECONCILIATION
|
||||
? "签收数量未确认或无效，国内整票签收事件未发布，须核对真实实签量后继续回写"
|
||||
: "原实发" + order.getPackageCount() + "箱，本次签收" + order.getSignedPackages()
|
||||
+ "箱，其余已退回；国内整票签收事件未发布，须按来源入仓单核对签收与退回处置后继续回写");
|
||||
pending.setCreatedAt(now); operations.insert(pending);
|
||||
}
|
||||
var completion = completeRelatedOrders(order, operatorId, now);
|
||||
deliveryCompletedOutbox.publish(order, completion.orders(), completion.proofs(), now.toInstant(ZoneOffset.UTC));
|
||||
customerVisits.ensureForCompletedDelivery(order, now);
|
||||
}
|
||||
Detail result = detail(order, false);
|
||||
@@ -166,20 +157,54 @@ public class DeliveryDocumentService {
|
||||
(int) categories.stream().filter(c -> !c.files().isEmpty()).count(), 3, categories, List.of("MANAGE_DOCUMENTS"), replayed);
|
||||
}
|
||||
private static long version(DeliveryOrder order) { return Objects.requireNonNullElse(order.getVersion(), 0L); }
|
||||
private List<LastlegOrder> completeRelatedOrders(DeliveryOrder delivery, LocalDateTime now) {
|
||||
List<LastlegOrder> completedOrders = lastlegOrders.selectList(new LambdaQueryWrapper<LastlegOrder>()
|
||||
.eq(LastlegOrder::getCustomerId, delivery.getCustomerId())
|
||||
.eq(LastlegOrder::getDeliveryOrderNo, delivery.getDeliveryNo()));
|
||||
for (LastlegOrder order : completedOrders) {
|
||||
// Preserve the outbound leg when receipt completion closes delivery.
|
||||
// Sub-order status is a composite fact; replacing it with DELIVERY alone
|
||||
// makes a successfully departed outbound appear unfinished or disappear.
|
||||
order.setSubOrderStatus(delivery.getSignedPackages().equals(delivery.getPackageCount())
|
||||
? "OUTBOUND:OUTBOUNDED;DELIVERY:DELIVERED" : "OUTBOUND:OUTBOUNDED;DELIVERY:PARTIALLY_RETURNED");
|
||||
order.setUpdatedAt(now);
|
||||
lastlegOrders.updateById(order);
|
||||
private record Completion(List<LastlegOrder> orders, Map<String, SourceDeliveryCompletionPolicy.Result> proofs) { }
|
||||
private Completion completeRelatedOrders(DeliveryOrder delivery, String operatorId, LocalDateTime now) {
|
||||
var related = new TreeMap<Long, LastlegOrder>();
|
||||
if (delivery.getDispatchOrderId() != null) {
|
||||
var ids = sourceEvidence.picks(delivery.getCustomerId(), delivery.getDispatchOrderId()).stream()
|
||||
.filter(p -> Objects.equals(p.groupKey(), delivery.getDeliveryGroupKey()))
|
||||
.map(SourceDeliveryCompletionPolicy.Pick::orderId).distinct().toList();
|
||||
if (!ids.isEmpty()) for (var tail : lastlegOrders.selectBatchIds(ids)) {
|
||||
if (!delivery.getCustomerId().equals(tail.getCustomerId())) throw conflict("派送来源客户归属不一致");
|
||||
related.put(tail.getId(), tail);
|
||||
}
|
||||
}
|
||||
return completedOrders;
|
||||
// Non-dispatch local workflows retain their established relation; never use this to prove a domestic source.
|
||||
for (var tail : lastlegOrders.selectList(new LambdaQueryWrapper<LastlegOrder>()
|
||||
.eq(LastlegOrder::getCustomerId, delivery.getCustomerId()).eq(LastlegOrder::getDeliveryOrderNo, delivery.getDeliveryNo()))) {
|
||||
if (!DeliveryCompletedOutboxService.domestic(tail)) related.putIfAbsent(tail.getId(),tail);
|
||||
else if (!related.containsKey(tail.getId())) pendingCompletion(delivery, operatorId, now,
|
||||
inboundLabel(tail) + "：缺少历史实际拣货来源，未发布整来源签收事件");
|
||||
}
|
||||
var sourceIds = related.values().stream().filter(DeliveryCompletedOutboxService::domestic)
|
||||
.map(LastlegOrder::getSourceInboundId).distinct().toList();
|
||||
var proofs = sourceIds.isEmpty() ? Map.<String,SourceDeliveryCompletionPolicy.Result>of()
|
||||
: sourceCompletion.proveAll(delivery.getCustomerId(),sourceIds);
|
||||
// All source locks are now held. Re-read current rows instead of overwriting with pre-lock snapshots.
|
||||
var current = new ArrayList<LastlegOrder>();
|
||||
for (long id : related.keySet()) {
|
||||
var tail = lastlegOrders.selectById(id);
|
||||
if (tail == null || !delivery.getCustomerId().equals(tail.getCustomerId())) throw conflict("派送来源已变化");
|
||||
if (DeliveryCompletedOutboxService.domestic(tail)) {
|
||||
var proof = proofs.get(tail.getSourceInboundId());
|
||||
if (proof != null && proof.fullySigned()) tail.setSubOrderStatus("OUTBOUND:OUTBOUNDED;DELIVERY:DELIVERED");
|
||||
else {
|
||||
tail.setSubOrderStatus("OUTBOUND:OUTBOUNDED;DELIVERY:WAIT_DELIVERY");
|
||||
pendingCompletion(delivery, operatorId, now, inboundLabel(tail) + "：来源签收未核齐；"
|
||||
+ (proof == null ? "缺少来源证明" : "已证明" + proof.provenSignedPackages() + "/" + proof.originalPackages() + "箱；" + (proof.unresolvedReasons().isEmpty() ? "仍有货物未完成签收" : "签收与退回来源证据尚未核齐，请核对实发、实收及签收记录")));
|
||||
}
|
||||
} else tail.setSubOrderStatus(Objects.equals(delivery.getSignedPackages(),delivery.getPackageCount())
|
||||
? "OUTBOUND:OUTBOUNDED;DELIVERY:DELIVERED" : "OUTBOUND:OUTBOUNDED;DELIVERY:PARTIALLY_RETURNED");
|
||||
tail.setUpdatedAt(now); lastlegOrders.updateById(tail); current.add(tail);
|
||||
}
|
||||
return new Completion(List.copyOf(current),proofs);
|
||||
}
|
||||
private static String inboundLabel(LastlegOrder order) { return order.getInboundCode()==null || order.getInboundCode().isBlank() ? "—" : order.getInboundCode(); }
|
||||
private void pendingCompletion(DeliveryOrder delivery, String operatorId, LocalDateTime now, String reason) {
|
||||
var operation = new DeliveryOperation(); operation.setCustomerId(delivery.getCustomerId());
|
||||
operation.setDeliveryOrderId(delivery.getId()); operation.setOperatorId(operatorId);
|
||||
operation.setOperationNode("FIRSTLEG_COMPLETION_PENDING"); operation.setResult(reason);
|
||||
operation.setCreatedAt(now); operations.insert(operation);
|
||||
}
|
||||
private void recordAutomaticCompletion(DeliveryOrder order, String operatorId, LocalDateTime now) {
|
||||
DeliveryOperation operation = new DeliveryOperation();
|
||||
|
||||
+84
@@ -0,0 +1,84 @@
|
||||
package com.baiyue.lastleg.delivery.service;
|
||||
|
||||
import com.baiyue.lastleg.delivery.mapper.DeliverySourceEvidenceMapper;
|
||||
import com.baiyue.lastleg.delivery.model.SourceDeliveryCompletionPolicy;
|
||||
import com.baiyue.lastleg.delivery.model.SourceDeliveryCompletionPolicy.*;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Propagation;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import java.util.*;
|
||||
|
||||
/** Caller locks its own delivery, then calls proveAll BEFORE any source-order updates.
|
||||
* Explicit READ_COMMITTED is required. Lock source orders in numeric ID order until commit;
|
||||
* never lock historical delivery/dispatch/exception/inventory rows from this query.
|
||||
* RC evidence reads see the preceding source-lock owner's committed signatures without waiting
|
||||
* for another transaction's delivery lock. All signature writers must follow this protocol.
|
||||
*/
|
||||
@Service
|
||||
public class SourceDeliveryCompletionQuery {
|
||||
private final DeliverySourceEvidenceMapper evidence;
|
||||
public SourceDeliveryCompletionQuery(DeliverySourceEvidenceMapper evidence) { this.evidence = evidence; }
|
||||
|
||||
/** Source uses UNIQUE(customer_id,source_region,source_inbound_id); mapper selectOne has no LIMIT.
|
||||
* A broken uniqueness invariant therefore raises TooManyResultsException, never picks one row.
|
||||
*/
|
||||
@Transactional(propagation = Propagation.MANDATORY)
|
||||
public Result prove(String customerId, String sourceInboundId) {
|
||||
return proveAll(customerId, List.of(sourceInboundId)).get(sourceInboundId);
|
||||
}
|
||||
|
||||
/** One call with ALL affected sources; do not loop prove() for a multi-source delivery. */
|
||||
@Transactional(propagation = Propagation.MANDATORY)
|
||||
public Map<String, Result> proveAll(String customerId, Collection<String> sourceInboundIds) {
|
||||
if (!TransactionSynchronizationManager.isActualTransactionActive()
|
||||
|| TransactionSynchronizationManager.isCurrentTransactionReadOnly())
|
||||
throw new IllegalStateException("Source signature proof requires the caller's active write transaction");
|
||||
if (!Objects.equals(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel(), TransactionDefinition.ISOLATION_READ_COMMITTED))
|
||||
throw new IllegalStateException("Source signature proof requires explicit READ_COMMITTED on the outer transaction");
|
||||
if (customerId == null || customerId.isBlank() || sourceInboundIds == null || sourceInboundIds.isEmpty()
|
||||
|| sourceInboundIds.stream().anyMatch(id -> id == null || id.isBlank()))
|
||||
throw new IllegalArgumentException("Customer and source inbound ID are required");
|
||||
var result = new LinkedHashMap<String, Result>(); var sources = new TreeMap<Long, Source>();
|
||||
for (String id : new TreeSet<>(sourceInboundIds)) {
|
||||
Source source = evidence.source(customerId, id);
|
||||
if (source == null) { result.put(id, new Result(id,0,0,false,List.of("SOURCE_NOT_FOUND","HISTORY_NOT_COMPLETE"))); continue; }
|
||||
if (!id.equals(source.sourceInboundId()) || source.orderId() <= 0 || sources.putIfAbsent(source.orderId(),source) != null)
|
||||
throw new IllegalStateException("Source evidence identity mismatch");
|
||||
}
|
||||
var locked = new ArrayList<Source>();
|
||||
for (var source : sources.values()) {
|
||||
Source current = evidence.lockSource(customerId, source.orderId());
|
||||
if (current == null || current.orderId() != source.orderId() || !current.sourceInboundId().equals(source.sourceInboundId()))
|
||||
throw new IllegalStateException("Source changed while acquiring completion lock");
|
||||
locked.add(current);
|
||||
}
|
||||
for (var source : locked) result.put(source.sourceInboundId(), proveLocked(customerId, source));
|
||||
return Collections.unmodifiableMap(result);
|
||||
}
|
||||
|
||||
private Result proveLocked(String customerId, Source source) {
|
||||
String sourceInboundId = source.sourceInboundId();
|
||||
var ids = Objects.requireNonNull(evidence.dispatchIds(customerId, source.orderId()), "Dispatch history missing");
|
||||
if (ids.stream().anyMatch(id -> id == null || id <= 0) || new HashSet<>(ids).size() != ids.size())
|
||||
throw new IllegalStateException("Invalid or duplicate historical dispatch IDs");
|
||||
var deliveries = new ArrayList<Delivery>(); var picks = new ArrayList<Pick>(); var returns = new ArrayList<Returned>();
|
||||
for (long id : ids.stream().sorted().toList()) {
|
||||
var dispatchDeliveries = Objects.requireNonNull(evidence.deliveries(customerId, id), "Delivery evidence missing");
|
||||
var dispatchPicks = Objects.requireNonNull(evidence.picks(customerId, id), "Pick evidence missing");
|
||||
var dispatchReturns = Objects.requireNonNull(evidence.returns(customerId, id), "Return evidence missing");
|
||||
if (dispatchDeliveries.isEmpty() || dispatchPicks.isEmpty()
|
||||
|| dispatchDeliveries.stream().anyMatch(row -> row == null || row.dispatchId() != id)
|
||||
|| dispatchPicks.stream().anyMatch(row -> row == null || row.dispatchId() != id)
|
||||
|| dispatchPicks.stream().noneMatch(row -> row.orderId() == source.orderId()))
|
||||
return new Result(sourceInboundId, source.packages(), 0, false,
|
||||
List.of("DISPATCH_HISTORY_INCOMPLETE:" + id, "HISTORY_NOT_COMPLETE"));
|
||||
var deliveryIds = new HashSet<Long>(); dispatchDeliveries.forEach(row -> deliveryIds.add(row.deliveryId()));
|
||||
if (dispatchReturns.stream().anyMatch(row -> row == null || !deliveryIds.contains(row.deliveryId())))
|
||||
throw new IllegalStateException("Return evidence belongs to another dispatch");
|
||||
deliveries.addAll(dispatchDeliveries); picks.addAll(dispatchPicks); returns.addAll(dispatchReturns);
|
||||
}
|
||||
return SourceDeliveryCompletionPolicy.evaluate(source, deliveries, picks, returns, true);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user