feat(appointments): support scoped receiving and stable trip continuation

This commit is contained in:
2026-10-16 15:43:04 +08:00
parent 4e2ea89403
commit 30f48bd926
84 changed files with 4879 additions and 136 deletions
@@ -3,7 +3,9 @@ package com.baiyue.lastleg.inboundplan.dto;
import java.util.List;
public record FbaAppointmentBatchContext(FbaAppointmentResponse appointment,List<SourceAllocation> sources,
List<String> allowedActions,List<String> blockReasons) {
List<String> allowedActions,List<String> blockReasons,
List<FbaAppointmentResponse> currentBatches,List<SourceRef> originalSourceRefs) {
public record SourceRef(String customerId,String sourceKey,String destinationKey) { }
public record SourceAllocation(String customerId,String sourceKey,String destinationKey,Integer packageCount,
Integer originalPackages,String sourceVersion,Integer cargoVerificationVersion,
List<String> shipmentReferences) { }
@@ -0,0 +1,13 @@
package com.baiyue.lastleg.inboundplan.receiving;
import java.util.List;
/** Original execution evidence, scoped by customer; work projections cannot authorize rebooking. */
public interface FbaAppointmentExecutionPort {
record Member(String batchKey,String customerId) { }
record Eligibility(Member member,boolean completed,boolean canRebook,String reason) { }
record Scope(String version,List<Eligibility> members) { }
Scope inspect(List<Member> members);
/** Locks original trips and selected customers' dispatch/execution rows before core batch locks. */
Scope lockAndInspect(List<Member> members,String expectedVersion);
}
@@ -0,0 +1,18 @@
package com.baiyue.lastleg.inboundplan.receiving;
import java.util.List;
/** Original Dispatch/Fleet authority; implementations never invent physical pallet splits. */
public interface FbaAppointmentTransportPort {
record Vehicle(String tripNo, Integer version, String dispatchNo, String customerId,
String batchKey, String status, int packageCount) { }
record Scope(String version, List<Vehicle> vehicles, List<String> blockReasons) { }
record Move(String batchKey, String customerId, String sourceKey, String destinationKey,
int originalPackages, int retainedPackages, String laterBatchKey) { }
record Result(List<Vehicle> vehicles, List<String> nextActions) { }
Scope inspect(List<String> batchKeys);
/** Lock order: trip -> dispatch -> outbound/PDA -> appointment; called BEFORE core locks batches. */
Scope lockAndInspect(List<String> batchKeys, String expectedVersion);
/** Same transaction as core allocation. Throw on failure so physical/allocation changes roll back together. */
Result apply(Scope locked, List<Move> moves, String operator, String reason);
}
@@ -0,0 +1,13 @@
package com.baiyue.lastleg.inboundplan.receiving;
import java.time.LocalDate;
import java.util.List;
/** Work domain owns the verified responsible person and durable follow-up inheritance. */
public interface FbaAppointmentWorkPort {
record Member(String batchKey,String customerId) { }
record Followup(String ownerId,String ownerName,LocalDate nextFollowupDate) { }
Followup resolveFollowup(String ownerId,LocalDate nextFollowupDate);
void seedFollowup(Member original,Member later,Followup followup,String operatorId);
void inheritFollowup(List<Member> originals,List<Member> successors,String operatorId);
}
@@ -0,0 +1,64 @@
package com.baiyue.lastleg.inboundplan.receiving;
import com.baiyue.lastleg.inboundplan.dto.FbaAppointmentDetails;
import com.baiyue.lastleg.inboundplan.dto.FbaAppointmentResponse;
import jakarta.validation.Valid;
import jakarta.validation.constraints.*;
import java.time.LocalDate;
import java.util.List;
/** Receiving coverage is separate from a stable cargo batch and from shared vehicles. */
public final class FbaReceivingContracts {
private FbaReceivingContracts() { }
public record Member(String batchKey, String appointmentNo, int batchVersion, String customerId,
List<Source> sources, String customerName) { }
public record Source(String sourceKey, String destinationKey, int packageCount,
String sourceVersion, Integer verificationVersion, String inboundCode) { }
public record Agreement(String agreementKey, int version, String status, String destinationKey,
FbaAppointmentDetails details, String timezone, List<Member> members,
String warehouseEvidence) { }
public record Excluded(Member member, String reason, String currentAgreementKey) { }
public record Pending(String changeKey, int version, String status, ChangeIntent intent,
List<Member> originalMembers, List<Agreement> originalAgreements,FbaAppointmentWorkPort.Followup followup) { }
public record CustomerPlan(String batchKey,String customerId,FbaAppointmentDetails details,String timezone) { }
public record BatchReceiving(String batchKey, List<Agreement> agreements, List<Pending> pendingChanges,List<CustomerPlan> plans) { }
public record Context(FbaAppointmentResponse batch, List<Agreement> agreements,
List<Pending> pendingChanges, FbaAppointmentTransportPort.Scope transport,
List<String> allowedActions, List<String> blockReasons,FbaAppointmentExecutionPort.Scope executionScope,List<CustomerPlan> plans) { }
public record GroupContext(Agreement agreement, List<Member> eligibleMembers, List<Excluded> excludedMembers,
FbaAppointmentTransportPort.Scope transport,FbaAppointmentExecutionPort.Scope executionScope) { }
public record Selection(@NotBlank String batchKey, @Min(0) int batchVersion,
@NotBlank String customerId) { }
public record Plan(@NotBlank @Size(max=64) String idempotencyKey,@NotNull @Valid Selection member,
@NotNull @Valid FbaAppointmentDetails details,@NotBlank String executionVersion) { }
public record ExecutionQuery(@NotEmpty @Size(max=200) List<FbaAppointmentExecutionPort.Member> members) { }
public record CargoVersion(@NotBlank String sourceKey,@NotBlank String destinationKey,
@NotBlank String expectedSourceVersion,@Min(1) Integer expectedVerificationVersion) { }
public record RefreshCargo(@NotBlank @Size(max=64) String idempotencyKey,@NotNull @Valid Selection member,
@NotEmpty @Size(max=100) List<@NotNull @Valid CargoVersion> sources,
@NotBlank String executionVersion) { }
public record Quantity(@NotBlank String sourceKey, @NotBlank String destinationKey,
@Min(0) int retainedPackages, @NotBlank String expectedSourceVersion,
Integer expectedVerificationVersion) { }
public record ChangeIntent(@NotBlank String batchKey, @Min(0) int batchVersion,
@NotBlank String customerId, @NotEmpty @Valid List<Quantity> quantities,
LocalDate laterDate, @NotBlank @Size(max=64) String ownerId,
LocalDate nextFollowupDate, @NotBlank @Size(max=1000) String reason,
@NotBlank String transportVersion) { }
public record Propose(@NotBlank @Size(max=64) String idempotencyKey, @NotNull @Valid ChangeIntent intent) { }
public record Resolve(@NotBlank @Size(max=64) String idempotencyKey, @Min(0) int version,
@NotBlank String decision, @Size(max=1000) String warehouseEvidence,
@NotBlank String transportVersion) { }
public record Booking(@NotEmpty @Valid List<Selection> members, @NotNull @Valid FbaAppointmentDetails details,
@NotBlank @Size(max=1000) String warehouseEvidence) { }
public record Book(@NotBlank @Size(max=64) String idempotencyKey, @NotEmpty @Valid List<Booking> bookings,
String cancelledAgreementKey, Integer cancelledAgreementVersion,
@NotBlank String executionVersion) { }
public record GroupChange(@NotBlank @Size(max=64) String idempotencyKey, @Min(0) int version,
@NotBlank String action, @Valid FbaAppointmentDetails details,
@NotBlank @Size(max=1000) String warehouseEvidence,
@NotBlank String transportVersion) { }
public record Step(String step, String status, String targetKey, String message) { }
public record Result(String changeKey, String status, List<FbaAppointmentResponse> batches,
List<Agreement> agreements, List<Step> steps, List<String> nextActions) { }
}
@@ -0,0 +1,52 @@
package com.baiyue.lastleg.inboundplan.receiving;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.security.AuthenticatedUser;
import jakarta.validation.Valid;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.security.core.annotation.AuthenticationPrincipal;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.server.ResponseStatusException;
import org.springframework.http.HttpStatus;
import static com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.*;
@RestController
@RequestMapping("/api/platform/v1/last-leg/appointment-receiving")
public class FbaReceivingController {
private static final String READ="hasAuthority('platform:appointment:list')";
private static final String EDIT="hasAuthority('platform:overseas-warehouse:inbound-order:fba-appointment')";
private static final String DELETE="hasAuthority('platform:appointment:delete')";
private final FbaReceivingService service;
public FbaReceivingController(FbaReceivingService service) {this.service=service;}
@GetMapping("/batches/{batchKey}") @PreAuthorize(READ)
public ApiResponse<Context> context(@AuthenticationPrincipal AuthenticatedUser user,@PathVariable String batchKey) {platform(user);return ApiResponse.success(service.context(batchKey));}
@GetMapping("/agreements/{agreementKey}") @PreAuthorize(READ)
public ApiResponse<GroupContext> group(@AuthenticationPrincipal AuthenticatedUser user,@PathVariable String agreementKey) {platform(user);return ApiResponse.success(service.groupContext(agreementKey));}
@PostMapping("/execution-scope") @PreAuthorize(READ)
public ApiResponse<FbaAppointmentExecutionPort.Scope> executionScope(@AuthenticationPrincipal AuthenticatedUser user,@Valid @RequestBody ExecutionQuery request) {platform(user);return ApiResponse.success(service.executionScope(request));}
@PostMapping("/changes") @PreAuthorize(EDIT+" and "+DELETE)
public ApiResponse<Result> propose(@AuthenticationPrincipal AuthenticatedUser user,@Valid @RequestBody Propose request) {return ApiResponse.success(service.propose(platform(user),request));}
@PostMapping("/changes/{changeKey}/resolve") @PreAuthorize(EDIT+" and "+DELETE)
public ApiResponse<Result> resolve(@AuthenticationPrincipal AuthenticatedUser user,@PathVariable String changeKey,@Valid @RequestBody Resolve request) {return ApiResponse.success(service.resolve(platform(user),changeKey,request));}
@PostMapping("/bookings") @PreAuthorize(EDIT)
public ApiResponse<Result> book(@AuthenticationPrincipal AuthenticatedUser user,@Valid @RequestBody Book request) {return ApiResponse.success(service.book(platform(user),request));}
@PostMapping("/plans") @PreAuthorize(EDIT)
public ApiResponse<Result> plan(@AuthenticationPrincipal AuthenticatedUser user,@Valid @RequestBody Plan request) {return ApiResponse.success(service.plan(platform(user),request));}
@PostMapping("/cargo-refresh") @PreAuthorize(EDIT)
public ApiResponse<Result> refreshCargo(@AuthenticationPrincipal AuthenticatedUser user,@Valid @RequestBody RefreshCargo request) {return ApiResponse.success(service.refreshCargo(platform(user),request));}
@PostMapping("/agreements/{agreementKey}/reschedule") @PreAuthorize(EDIT)
public ApiResponse<Result> reschedule(@AuthenticationPrincipal AuthenticatedUser user,@PathVariable String agreementKey,@Valid @RequestBody GroupChange request) {
var operator=platform(user);if(!"RESCHEDULE".equals(request.action())) throw new ResponseStatusException(HttpStatus.BAD_REQUEST,"此入口只允许改期");
return ApiResponse.success(service.changeGroup(operator,agreementKey,request));
}
@PostMapping("/agreements/{agreementKey}/cancel") @PreAuthorize(DELETE)
public ApiResponse<Result> cancel(@AuthenticationPrincipal AuthenticatedUser user,@PathVariable String agreementKey,@Valid @RequestBody GroupChange request) {
var operator=platform(user);if(!"CANCEL".equals(request.action())) throw new ResponseStatusException(HttpStatus.BAD_REQUEST,"此入口只允许取消");
return ApiResponse.success(service.changeGroup(operator,agreementKey,request));
}
private static String platform(AuthenticatedUser user) {
if(user==null || user.userId()==null) throw new ResponseStatusException(HttpStatus.UNAUTHORIZED);
if(!"PLATFORM".equals(user.appCode()) || !Long.valueOf(0).equals(user.tenantId())) throw new ResponseStatusException(HttpStatus.FORBIDDEN);
return String.valueOf(user.userId());
}
}
@@ -0,0 +1,66 @@
package com.baiyue.lastleg.inboundplan.receiving;
import org.apache.ibatis.annotations.*;
import java.util.List;
@Mapper
public interface FbaReceivingMapper {
record GroupRow(String agreementKey, int version, String status, String destinationKey, String confirmationNo, String snapshotJson) { }
record ChangeRow(String changeKey, int version, String status, String batchKey, String customerId, String snapshotJson, String resultJson) { }
record BatchJson(String batchKey,String snapshotJson) { }
record CustomerLabel(String customerId,String customerName) { }
record PlanRow(String batchKey,String customerId,String snapshotJson) { }
@Select("<script>SELECT p.batch_key,p.customer_id,p.snapshot_json FROM ll_fba_receiving_plan p JOIN ll_inbound_fba_appointment a ON p.batch_key=CONCAT('FBA_APPOINTMENT:',a.id) WHERE a.replacement_nos_json IS NULL AND EXISTS (SELECT 1 FROM ll_inbound_fba_appointment_item i WHERE i.appointment_id=a.id AND i.customer_id=p.customer_id AND i.active=1) AND p.batch_key IN <foreach collection='keys' item='key' open='(' separator=',' close=')'>#{key}</foreach> ORDER BY p.batch_key,p.customer_id</script>")
List<PlanRow> plans(@Param("keys") List<String> keys);
@Delete("DELETE FROM ll_fba_receiving_plan WHERE batch_key=#{batch} AND customer_id=#{customer}")
int deletePlan(@Param("batch") String batch,@Param("customer") String customer);
@Insert("INSERT INTO ll_fba_receiving_plan(batch_key,customer_id,snapshot_json) VALUES(#{batchKey},#{customerId},#{snapshotJson})")
int insertPlan(PlanRow row);
@Select("SELECT DISTINCT i.customer_id,o.customer_name FROM ll_inbound_fba_appointment_item i JOIN ll_order o ON o.id=i.order_id AND o.customer_id=i.customer_id WHERE CONCAT('FBA_APPOINTMENT:',i.appointment_id)=#{batch} AND i.active=1")
List<CustomerLabel> labels(String batch);
record LiveSource(String customerId,String sourceKey,String destinationKey,Integer packageCount) { }
@Select("SELECT customer_id,source_key,CONCAT('FBA:',fba_warehouse_code) AS destination_key,package_count FROM ll_inbound_fba_appointment_item WHERE CONCAT('FBA_APPOINTMENT:',appointment_id)=#{batch} AND active=1")
List<LiveSource> liveSources(String batch);
@Select("<script>SELECT DISTINCT m.batch_key,a.snapshot_json FROM ll_fba_receiving_member m JOIN ll_fba_receiving_agreement a ON a.agreement_key=m.agreement_key WHERE m.active_ref IS NOT NULL AND m.batch_key IN <foreach collection='keys' item='key' open='(' separator=',' close=')'>#{key}</foreach> ORDER BY m.batch_key</script>")
List<BatchJson> currentGroups(@Param("keys") List<String> keys);
@Select("<script>SELECT batch_key,snapshot_json FROM ll_fba_receiving_change WHERE status='WAITING' AND batch_key IN <foreach collection='keys' item='key' open='(' separator=',' close=')'>#{key}</foreach> ORDER BY batch_key,change_key</script>")
List<BatchJson> pendingGroups(@Param("keys") List<String> keys);
@Select("SELECT agreement_key,version,status,destination_key,confirmation_no,snapshot_json FROM ll_fba_receiving_agreement WHERE agreement_key=#{key}")
GroupRow group(String key);
@Select("SELECT agreement_key,version,status,destination_key,confirmation_no,snapshot_json FROM ll_fba_receiving_agreement WHERE agreement_key=#{key} FOR UPDATE")
@Options(flushCache=Options.FlushCachePolicy.TRUE,useCache=false)
GroupRow lockGroup(String key);
@Insert("INSERT INTO ll_fba_receiving_agreement(agreement_key,version,status,destination_key,confirmation_no,snapshot_json) VALUES(#{agreementKey},#{version},#{status},#{destinationKey},#{confirmationNo},#{snapshotJson})")
int insertGroup(GroupRow row);
@Update("UPDATE ll_fba_receiving_agreement SET version=#{version},status=#{status},confirmation_no=#{confirmationNo},snapshot_json=#{snapshotJson} WHERE agreement_key=#{agreementKey}")
int updateGroup(GroupRow row);
@Insert("INSERT INTO ll_fba_receiving_revision(agreement_key,version,snapshot_json) VALUES(#{agreementKey},#{version},#{snapshotJson})")
int revision(GroupRow row);
@Select("SELECT DISTINCT agreement_key FROM ll_fba_receiving_member WHERE batch_key=#{batch} ORDER BY agreement_key")
List<String> groupKeys(String batch);
@Select("SELECT agreement_key FROM ll_fba_receiving_member WHERE batch_key=#{batch} AND customer_id=#{customer} AND active_ref IS NOT NULL")
String currentKey(@Param("batch") String batch,@Param("customer") String customer);
@Insert("INSERT INTO ll_fba_receiving_member(agreement_key,batch_key,customer_id,active_ref) VALUES(#{group},#{batch},#{customer},#{activeRef})")
int addMember(@Param("group") String group,@Param("batch") String batch,@Param("customer") String customer,@Param("activeRef") String activeRef);
@Update("UPDATE ll_fba_receiving_member SET active_ref=NULL WHERE agreement_key=#{group}")
int releaseGroup(String group);
@Update("UPDATE ll_fba_receiving_member SET active_ref=NULL WHERE agreement_key=#{group} AND batch_key=#{batch} AND customer_id=#{customer}")
int releaseMember(@Param("group") String group,@Param("batch") String batch,@Param("customer") String customer);
@Select("SELECT change_key,version,status,batch_key,customer_id,snapshot_json,result_json FROM ll_fba_receiving_change WHERE change_key=#{key}")
ChangeRow change(String key);
@Select("SELECT change_key,version,status,batch_key,customer_id,snapshot_json,result_json FROM ll_fba_receiving_change WHERE change_key=#{key} FOR UPDATE")
@Options(flushCache=Options.FlushCachePolicy.TRUE,useCache=false)
ChangeRow lockChange(String key);
@Select("SELECT change_key,version,status,batch_key,customer_id,snapshot_json,result_json FROM ll_fba_receiving_change WHERE batch_key=#{batch} AND status='WAITING' ORDER BY change_key")
List<ChangeRow> pending(String batch);
@Select("SELECT result_json FROM ll_fba_receiving_change WHERE batch_key=#{batch} AND status='APPLIED' ORDER BY change_key")
List<String> appliedResults(String batch);
@Select("SELECT snapshot_json FROM ll_fba_receiving_change WHERE batch_key=#{batch} AND status='APPLIED' ORDER BY change_key")
List<String> appliedChanges(String batch);
@Select("SELECT COUNT(*) FROM ll_fba_receiving_change WHERE batch_key=#{batch} AND customer_id=#{customer} AND status='WAITING'")
long pendingCount(@Param("batch") String batch,@Param("customer") String customer);
@Insert("INSERT INTO ll_fba_receiving_change(change_key,version,status,batch_key,customer_id,snapshot_json,active_ref) VALUES(#{changeKey},#{version},#{status},#{batchKey},#{customerId},#{snapshotJson},CONCAT(#{batchKey},'/',#{customerId}))")
int insertChange(ChangeRow row);
@Update("UPDATE ll_fba_receiving_change SET version=#{version},status=#{status},snapshot_json=#{snapshotJson},result_json=#{resultJson},active_ref=NULL WHERE change_key=#{changeKey}")
int finishChange(ChangeRow row);
}
@@ -0,0 +1,284 @@
package com.baiyue.lastleg.inboundplan.receiving;
import com.baiyue.lastleg.inboundplan.dto.FbaAppointmentResponse;
import com.baiyue.lastleg.inboundplan.mapper.FbaAppointmentCommandMapper;
import com.baiyue.lastleg.inboundplan.service.InboundFbaAppointmentService;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.*;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.http.HttpStatus;
import org.springframework.web.server.ResponseStatusException;
import java.time.*;
import java.util.*;
import java.util.function.Supplier;
import static com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.*;
import static com.baiyue.lastleg.inboundplan.receiving.FbaReceivingStore.conflict;
@Service
@Transactional(isolation=Isolation.READ_COMMITTED)
public class FbaReceivingService {
private final FbaReceivingStore store;
private final InboundFbaAppointmentService batches;
private final FbaAppointmentTransportPort transport;
private final FbaAppointmentCommandMapper commands;
private final FbaAppointmentWorkPort work;
private final FbaAppointmentExecutionPort execution;
public FbaReceivingService(FbaReceivingStore store,InboundFbaAppointmentService batches,FbaAppointmentTransportPort transport,FbaAppointmentCommandMapper commands,FbaAppointmentWorkPort work,FbaAppointmentExecutionPort execution) {
this.store=store;this.batches=batches;this.transport=transport;this.commands=commands;this.work=work;this.execution=execution;
}
@Transactional(readOnly=true)
public FbaAppointmentExecutionPort.Scope executionScope(ExecutionQuery request) { return execution.inspect(request.members()); }
@Transactional(readOnly=true)
public Context context(String key) {
var batch=batches.batchSnapshots(List.of(key)).get(0);var pending=store.pending(key);
var agreements=store.agreements(key);
var scope=transport.inspect(List.of(key));var blocks=new ArrayList<>(scope.blockReasons());
if(!batch.replacementAppointmentNos().isEmpty()) blocks.add("原批次已全部转入后续安排,请进入当前批次");
var actions=new ArrayList<String>();
var customers=batch.items().stream().map(FbaAppointmentResponse.Item::customerId).collect(java.util.stream.Collectors.toSet());
var covered=agreements.stream().filter(a->"ACTIVE".equals(a.status())).flatMap(a->a.members().stream()).filter(m->key.equals(m.batchKey())).map(Member::customerId).collect(java.util.stream.Collectors.toSet());
var waiting=pending.stream().map(p->p.intent().customerId()).collect(java.util.stream.Collectors.toSet());
var executionScope=execution.inspect(customers.stream().sorted().map(c->new FbaAppointmentExecutionPort.Member(key,c)).toList());
if(blocks.isEmpty()) {
if(customers.stream().anyMatch(c->!waiting.contains(c))) actions.add("PROPOSE_CUSTOMER_CHANGE");
}
if(batch.replacementAppointmentNos().isEmpty() && executionScope.members().stream().anyMatch(e->e.canRebook() && !e.completed() && !covered.contains(e.member().customerId()) && !waiting.contains(e.member().customerId()))) {
actions.add("BOOK_RECEIVING");actions.add("PLAN_RECEIVING");actions.add("REFRESH_CARGO");
}
return new Context(batch,agreements,pending,scope,List.copyOf(actions),List.copyOf(blocks),executionScope,store.plans(key));
}
@Transactional(readOnly=true)
public GroupContext groupContext(String key) {
var group=store.agreement(key,false);var eligible=new ArrayList<Member>();var excluded=new ArrayList<Excluded>();
var executionScope=execution.inspect(group.members().stream().map(m->new FbaAppointmentExecutionPort.Member(m.batchKey(),m.customerId())).toList());
for(var old:group.members()) {
var snapshot=batches.batchSnapshots(List.of(old.batchKey())).get(0);
var current=store.currentAgreement(old.batchKey(),old.customerId());
var member=snapshot.replacementAppointmentNos().isEmpty()?store.members(snapshot).stream().filter(m->m.customerId().equals(old.customerId())).findFirst().orElse(null):null;
String reason=!snapshot.replacementAppointmentNos().isEmpty()?"原批次已转入其他批次":member==null?"客户已移出本批":
store.hasPending(old.batchKey(),old.customerId())?"正在处理其他待确认变化":
current!=null && !current.agreementKey().equals(key)?"已另行预约":null;
var eligibility=executionScope.members().stream().filter(e->e.member().batchKey().equals(old.batchKey()) && e.member().customerId().equals(old.customerId())).findFirst().orElseThrow(()->conflict("客户执行资格读取不完整"));
if(reason==null && (eligibility.completed() || !eligibility.canRebook())) reason=eligibility.completed()?"本客户货物已实际履约完成":eligibility.reason();
if(reason==null && !eligibility.canRebook()) reason="本客户已开始执行,请从原运输异常处理";
if(reason==null) eligible.add(member);else excluded.add(new Excluded(old,reason,current==null?null:current.agreementKey()));
}
var transportMembers="CANCELLED".equals(group.status()) && !eligible.isEmpty()?eligible:group.members();
var scope=transport.inspect(transportMembers.stream().map(Member::batchKey).distinct().sorted().toList());
// Executed authority may block all further ordinary changes; no synthetic completion here.
var bookingScope=execution.inspect(eligible.stream().map(m->new FbaAppointmentExecutionPort.Member(m.batchKey(),m.customerId())).toList());
return new GroupContext(group,List.copyOf(eligible),List.copyOf(excluded),scope,bookingScope);
}
public Result propose(String operator,Propose request) {
return command("RECEIVING_PROPOSE",operator,request.idempotencyKey(),request,()->{
var intent=request.intent();var followup=work.resolveFollowup(intent.ownerId(),intent.nextFollowupDate());
var scope=transport.lockAndInspect(List.of(intent.batchKey()),intent.transportVersion());requireMutable(scope);
var batch=batches.lockReceivingBatch(intent.batchKey());requireVersion(batch,intent.batchVersion());
recheckTransport(List.of(intent.batchKey()),scope);
if(store.hasPending(intent.batchKey(),intent.customerId())) throw conflict("本客户已有待确认变化,请继续原记录");
var member=store.members(batch).stream().filter(m->m.customerId().equals(intent.customerId())).findFirst().orElseThrow(()->bad("客户不属于本批次"));
validateQuantities(member,intent);
var group=store.currentAgreement(batch.batchKey(),intent.customerId());
if(group!=null) {group=store.agreement(group.agreementKey(),true);requireActive(group);}
var pending=new Pending("CH-"+UUID.randomUUID().toString().replace("-","").toUpperCase(Locale.ROOT),0,"WAITING",intent,List.of(member),group==null?List.of():List.of(group),followup);
store.save(pending);
batches.receivingChanged(List.of(batch));
return new Result(pending.changeKey(),"WAITING",List.of(batch),pending.originalAgreements(),List.of(new Step("RECORD_CHANGE","SUCCESS",pending.changeKey(),"原安排及数量保留,待确认变化已阻止相关发车")),List.of("CONTINUE_WAREHOUSE_CONFIRMATION"));
});
}
public Result resolve(String operator,String changeKey,Resolve request) {
return command("RECEIVING_RESOLVE:"+changeKey,operator,request.idempotencyKey(),request,()->{
var initial=store.change(changeKey,false);var intent=initial.intent();
// Withdrawing only removes the pending proposal. It must remain possible after
// picking starts, without editing cargo, the appointment, or vehicle facts.
var withdrawing="WITHDRAW".equals(request.decision());
var followup=withdrawing?null:work.resolveFollowup(intent.ownerId(),intent.nextFollowupDate());
var scope=withdrawing?null:transport.lockAndInspect(List.of(intent.batchKey()),request.transportVersion());
if(!withdrawing) requireMutable(scope);
var batch=batches.lockReceivingBatch(intent.batchKey());var pending=store.change(changeKey,true);
if(pending.version()!=request.version() || !"WAITING".equals(pending.status())) throw conflict("变化已被处理,请重新读取当前结果");
if(withdrawing) {
var result=new Result(changeKey,"WITHDRAWN",List.of(batch),pending.originalAgreements(),List.of(new Step("WITHDRAW_CHANGE","SUCCESS",changeKey,"变化已撤回,原安排保留")),List.of("REVIEW_TRANSPORT"));
store.finish(pending,"WITHDRAWN",result);batches.receivingChanged(List.of(batch));return result;
}
if(!"CONFIRM".equals(request.decision()) || blank(request.warehouseEvidence())) throw bad("请填写仓库明确同意保留量及移出量的回复依据");
requireVersion(batch,intent.batchVersion());
Agreement original=null;
for(var expected:pending.originalAgreements()) {
var current=store.agreement(expected.agreementKey(),true);
if(current.version()!=expected.version() || !"ACTIVE".equals(current.status())) throw conflict("仓库预约覆盖已变化,不能沿旧回复处理");
original=current;
}
var currentKey=store.currentAgreement(intent.batchKey(),intent.customerId());
if(!Objects.equals(currentKey==null?null:currentKey.agreementKey(),original==null?null:original.agreementKey())) throw conflict("本客户有效预约关系已变化");
var changed=batches.moveReceivingCustomer(operator,intent);var retained=changed.get(0);var later=changed.get(1);
work.seedFollowup(new FbaAppointmentWorkPort.Member(intent.batchKey(),intent.customerId()),new FbaAppointmentWorkPort.Member(later.batchKey(),intent.customerId()),followup,operator);
var movements=new ArrayList<FbaAppointmentTransportPort.Move>();
for(var source:pending.originalMembers().get(0).sources()) {
var q=intent.quantities().stream().filter(v->v.sourceKey().equals(source.sourceKey()) && v.destinationKey().equals(source.destinationKey())).findFirst().orElseThrow();
movements.add(new FbaAppointmentTransportPort.Move(intent.batchKey(),intent.customerId(),source.sourceKey(),source.destinationKey(),source.packageCount(),q.retainedPackages(),later.batchKey()));
}
var physical=transport.apply(scope,List.copyOf(movements),operator,intent.reason());
var agreements=new ArrayList<Agreement>();
if(original!=null) {
var members=new ArrayList<>(original.members());members.removeIf(m->m.batchKey().equals(intent.batchKey()) && m.customerId().equals(intent.customerId()));
var remaining=store.members(retained).stream().filter(m->m.customerId().equals(intent.customerId())).findFirst();
if(remaining.isPresent()) members.add(remaining.get());else store.releaseMember(original.agreementKey(),intent.batchKey(),intent.customerId());
agreements.add(members.isEmpty()?store.cancel(original,"SUPERSEDED",request.warehouseEvidence()):store.update(original,members,original.details(),request.warehouseEvidence()));
}
var projected=retained.items().isEmpty()?retained:batches.projectReceiving(intent.batchKey(),false);
var steps=List.of(new Step("ALLOCATE_CARGO","SUCCESS",later.batchKey(),"原来源总量不变,后送量已落在新批次"),new Step("ADJUST_TRANSPORT","SUCCESS",intent.batchKey(),"原车辆按实际保留货物调整"));
var actions=new ArrayList<String>();actions.add("BOOK_LATER_BATCH");actions.addAll(physical.nextActions());
var result=new Result(changeKey,"APPLIED",List.of(projected,later),List.copyOf(agreements),steps,List.copyOf(actions));
store.finish(pending,"APPLIED",result);batches.receivingChanged(result.batches());return result;
});
}
public Result refreshCargo(String operator,RefreshCargo request) {
return command("RECEIVING_REFRESH_CARGO",operator,request.idempotencyKey(),request,()->{
var selection=request.member();var selected=List.of(new FbaAppointmentExecutionPort.Member(selection.batchKey(),selection.customerId()));
var locked=execution.lockAndInspect(selected,request.executionVersion());requireRebookable(selected,locked);
var batch=batches.lockReceivingBatch(selection.batchKey());requireVersion(batch,selection.batchVersion());
var current=execution.inspect(selected);
if(!Objects.equals(locked.version(),current.version())) throw conflict("客户执行关联已变化,请重新核对当前范围");
requireRebookable(selected,current);
if(store.hasPending(selection.batchKey(),selection.customerId())) throw conflict("本客户有待确认变化,请先处理或撤回原变化再核对资料");
if(store.currentAgreement(selection.batchKey(),selection.customerId())!=null) throw conflict("本客户已有正式确认,不能直接采用新货物资料,请先处理其有效预约");
var refreshed=batches.refreshReceivingCargo(selection.batchKey(),selection.customerId(),request.sources());
boolean changed=refreshed.version()!=batch.version();
return new Result(null,changed?"APPLIED":"UNCHANGED",List.of(refreshed),List.of(),changed?List.of(new Step("REFRESH_CARGO","SUCCESS",batch.batchKey(),"本客户已采用核对后的最新资料,批次数量及其他客户安排保留")):List.of(),List.of("CONTINUE_CUSTOMER_PLAN"));
});
}
public Result plan(String operator,Plan request) {
return command("RECEIVING_PLAN",operator,request.idempotencyKey(),request,()->{
var selection=request.member();var memberKey=new FbaAppointmentExecutionPort.Member(selection.batchKey(),selection.customerId());
var selected=List.of(memberKey);var locked=execution.lockAndInspect(selected,request.executionVersion());requireRebookable(selected,locked);
var batch=batches.lockReceivingBatch(selection.batchKey());requireVersion(batch,selection.batchVersion());
var current=execution.inspect(selected);
if(!Objects.equals(locked.version(),current.version())) throw conflict("客户执行关联已变化,请重新核对当前范围");
requireRebookable(selected,current);
if(store.hasPending(selection.batchKey(),selection.customerId())) throw conflict("本客户有待确认变化,请先处理原变化");
if(store.currentAgreement(selection.batchKey(),selection.customerId())!=null) throw conflict("本客户已有正式确认,请使用明确范围的改期或取消操作");
var member=store.member(batch,selection.customerId());if(member==null) throw conflict("客户已移出原批次");
var timezone=batches.validateReceivingPlan(member,request.details());
var plan=new CustomerPlan(batch.batchKey(),selection.customerId(),FbaReceivingStore.normalized(request.details()),timezone);
if(plan.equals(store.plan(batch.batchKey(),selection.customerId()))) return new Result(null,"UNCHANGED",List.of(batch),List.of(),List.of(),List.of("CONTINUE_CUSTOMER_PLAN"));
store.freezeUnconfirmedPlans(batch);
store.savePlan(plan);var projected=batches.projectReceiving(batch.batchKey(),false);
return new Result(null,"APPLIED",List.of(projected),List.of(),List.of(new Step("PLAN_RECEIVING","SUCCESS",batch.batchKey(),"仅本客户计划已保存,其他客户安排及货量保留")),List.of("CONTINUE_CUSTOMER_PLAN"));
});
}
public Result book(String operator,Book request) {
return command("RECEIVING_BOOK",operator,request.idempotencyKey(),request,()->{
if(request.bookings()==null || request.bookings().isEmpty() || request.bookings().size()>100) throw bad("请选择需要重新预约的客户");
var selections=request.bookings().stream().flatMap(b->b.members().stream()).toList();
var keys=selections.stream().map(Selection::batchKey).distinct().sorted().toList();
var executionMembers=selections.stream().map(s->new FbaAppointmentExecutionPort.Member(s.batchKey(),s.customerId())).distinct().toList();
var executionScope=execution.lockAndInspect(executionMembers,request.executionVersion());
requireRebookable(executionMembers,executionScope);
var snapshots=lock(keys);
// An unbound customer can acquire a new dispatch before we obtain the batch lock.
// Binding shares that batch lock, so rereading here closes the gap without reversing locks.
var currentExecution=execution.inspect(executionMembers);
if(!Objects.equals(executionScope.version(),currentExecution.version())) throw conflict("客户执行关联已变化,请重新核对当前范围");
requireRebookable(executionMembers,currentExecution);
snapshots.values().forEach(store::freezeUnconfirmedPlans);
Agreement cancelled=null;
if(request.cancelledAgreementKey()!=null) {
cancelled=store.agreement(request.cancelledAgreementKey(),true);
if(!"CANCELLED".equals(cancelled.status()) || !Objects.equals(cancelled.version(),request.cancelledAgreementVersion())) throw conflict("取消组版本已变化,请重新读取待重约范围");
}
var seen=new HashSet<String>();var saved=new ArrayList<Agreement>();
for(var booking:request.bookings()) {
if(blank(booking.warehouseEvidence()) || booking.members()==null || booking.members().isEmpty()) throw bad("请记录仓库确认覆盖本次客户和货量的依据");
var members=new ArrayList<Member>();
for(var selection:booking.members()) {
if(!seen.add(selection.batchKey()+"/"+selection.customerId())) throw bad("同客户批次不能重复预约");
var batch=snapshots.get(selection.batchKey());requireVersion(batch,selection.batchVersion());
if(store.hasPending(selection.batchKey(),selection.customerId())) throw conflict("该客户有其他待确认变化,不能直接重约");
if(store.currentAgreement(selection.batchKey(),selection.customerId())!=null) throw conflict("该客户已有有效收货预约,请明确变更当前预约范围");
if(cancelled!=null && cancelled.members().stream().noneMatch(m->m.batchKey().equals(selection.batchKey()) && m.customerId().equals(selection.customerId()))) throw conflict("所选客户不属于该取消组的原覆盖范围");
members.add(store.members(batch).stream().filter(m->m.customerId().equals(selection.customerId())).findFirst().orElseThrow(()->conflict("客户已移出原批次")));
}
var timezone=batches.validateReceiving(members,booking.details());saved.add(store.create(members,booking.details(),timezone,booking.warehouseEvidence()));
members.forEach(m->store.deletePlan(m.batchKey(),m.customerId()));
}
var projected=keys.stream().map(k->batches.projectReceiving(k,false)).toList();
return new Result(null,"APPLIED",projected,List.copyOf(saved),saved.stream().map(a->new Step("BOOK_RECEIVING","SUCCESS",a.agreementKey(),"使用原批次形成新的有效收货预约" )).toList(),List.of("REVIEW_TRANSPORT"));
});
}
public Result changeGroup(String operator,String key,GroupChange request) {
return command("RECEIVING_GROUP:"+key,operator,request.idempotencyKey(),request,()->{
var initial=store.agreement(key,false);var keys=initial.members().stream().map(Member::batchKey).distinct().sorted().toList();
var scope=transport.lockAndInspect(keys,request.transportVersion());requireMutable(scope);var snapshots=lock(keys);
recheckTransport(keys,scope);
var group=store.agreement(key,true);requireActive(group);
if(group.version()!=request.version()) throw conflict("共同预约范围已变化,请重新核对全部客户");
var members=new ArrayList<Member>();
for(var old:group.members()) {
if(store.hasPending(old.batchKey(),old.customerId())) throw conflict("当前成员有待确认变化,请先处理其实际范围");
var current=store.currentAgreement(old.batchKey(),old.customerId());
if(current==null || !key.equals(current.agreementKey())) throw conflict("成员已退出本预约,请重新读取范围");
members.add(store.members(snapshots.get(old.batchKey())).stream().filter(m->m.customerId().equals(old.customerId())).findFirst().orElseThrow(()->conflict("当前覆盖货物已变化")));
}
if(blank(request.warehouseEvidence())) throw bad("请记录本次仓库回复依据");
Agreement changed;
if("CANCEL".equals(request.action())) {
changed=store.cancel(group,"CANCELLED",request.warehouseEvidence());
members.forEach(m->store.deletePlan(m.batchKey(),m.customerId()));
}
else if("RESCHEDULE".equals(request.action())) {
batches.validateReceiving(members,request.details());
if(Objects.equals(group.details(),request.details())) return new Result(null,"UNCHANGED",List.copyOf(snapshots.values()),List.of(group),List.of(),List.of("REVIEW_TRANSPORT"));
changed=store.update(group,members,request.details(),request.warehouseEvidence());
if(changed.equals(group)) return new Result(null,"UNCHANGED",List.copyOf(snapshots.values()),List.of(group),List.of(),List.of("REVIEW_TRANSPORT"));
} else throw bad("请选择整份改期或整份取消");
var projected=keys.stream().map(k->batches.projectReceiving(k,"CANCEL".equals(request.action()))).toList();
return new Result(null,"APPLIED",projected,List.of(changed),List.of(new Step(request.action(),"SUCCESS",key,"按服务端当前覆盖成员处理,车辆和批次数量保留")),List.of("CANCEL".equals(request.action())?"REBOOK_GROUP":"REVIEW_TRANSPORT"));
});
}
private Map<String,FbaAppointmentResponse> lock(List<String> keys) {var result=new LinkedHashMap<String,FbaAppointmentResponse>();for(var key:keys) result.put(key,batches.lockReceivingBatch(key));return result;}
private void recheckTransport(List<String> keys,FbaAppointmentTransportPort.Scope locked) {
// Original bindings also lock the batch. Re-read after acquiring it to reject a new
// association created between the initial transport locks and the batch lock.
var current=transport.inspect(keys);
if(!Objects.equals(locked.version(),current.version())) throw conflict("运输关联已变化,请重新核对当前范围");
requireMutable(current);
}
private static void requireRebookable(List<FbaAppointmentExecutionPort.Member> members,FbaAppointmentExecutionPort.Scope scope) {
if(scope==null || scope.members()==null || !new HashSet<>(members).equals(scope.members().stream().map(FbaAppointmentExecutionPort.Eligibility::member).collect(java.util.stream.Collectors.toSet()))) throw conflict("客户执行资格读取不完整");
for(var member:scope.members()) if(member.completed() || !member.canRebook()) throw conflict(member.completed()?"已完成货物不能重新预约":blank(member.reason())?"本客户已开始执行,请从原运输异常处理":member.reason());
}
private static void validateQuantities(Member member,ChangeIntent intent) {
if(intent.quantities()==null || intent.quantities().size()!=member.sources().size()) throw bad("请填写本客户每票原来源的本次先送箱数");
var seen=new HashSet<String>();boolean changed=false;
for(var q:intent.quantities()) {
if(!seen.add(q.sourceKey()+"/"+q.destinationKey())) throw bad("来源不可重复");
var source=member.sources().stream().filter(s->s.sourceKey().equals(q.sourceKey()) && s.destinationKey().equals(q.destinationKey())).findFirst().orElseThrow(()->bad("所选来源不属于本客户批次"));
if(q.retainedPackages()<0 || q.retainedPackages()>source.packageCount()) throw bad("本次先送量须在零至本批箱数之间");
if(!Objects.equals(q.expectedSourceVersion(),source.sourceVersion()) || !Objects.equals(q.expectedVerificationVersion(),source.verificationVersion())) throw conflict("来源或核实版本已变化");
changed|=q.retainedPackages()!=source.packageCount();
}
if(!changed) throw bad("本次货量没有变化,无需保存调整");
}
private Result command(String operation,String operator,String key,Object request,Supplier<Result> action) {
if(blank(operator)||blank(key)||key.length()>64) throw bad("操作人及请求标识不能为空");
String hash;
try {hash=HexFormat.of().formatHex(java.security.MessageDigest.getInstance("SHA-256").digest((operation+"\n"+operator+"\n"+store.write(request)).getBytes(java.nio.charset.StandardCharsets.UTF_8)));}catch(Exception e){throw new IllegalStateException(e);}
var existing=commands.find(key);if(existing!=null) return replay(existing,hash);
try {commands.begin(key,hash,operation,operator,LocalDateTime.now(ZoneOffset.UTC));}catch(DuplicateKeyException duplicate){return replay(commands.find(key),hash);}
try {
var result=action.get();
if(commands.finish(key,result.batches().isEmpty()?null:result.batches().get(0).appointmentNo(),store.write(result))!=1) throw conflict("命令结果未保存");
return result;
} catch(DuplicateKeyException duplicate) {throw conflict("当前目的仓约号或客户覆盖已被其他预约使用,请重新读取范围");}
}
private Result replay(FbaAppointmentCommandMapper.Command row,String hash) {
if(row==null || !hash.equals(row.requestHash()) || row.responseJson()==null) throw conflict("同一请求标识的内容不同或结果尚未就绪");
return store.read(row.responseJson(),Result.class);
}
private static void requireVersion(FbaAppointmentResponse batch,int version) {if(batch.version()!=version) throw conflict("批次已更新,请刷新后重新核对");}
private static void requireActive(Agreement group) {if(!"ACTIVE".equals(group.status())) throw conflict("原收货预约已失效,请使用重新预约入口");}
private static void requireMutable(FbaAppointmentTransportPort.Scope scope) {if(!scope.blockReasons().isEmpty()) throw conflict(String.join(";",scope.blockReasons()));}
private static boolean blank(String s) {return s==null || s.isBlank();}
private static ResponseStatusException bad(String message) {return new ResponseStatusException(HttpStatus.BAD_REQUEST,message);}
}
@@ -0,0 +1,147 @@
package com.baiyue.lastleg.inboundplan.receiving;
import com.baiyue.lastleg.inboundplan.dto.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.stereotype.Component;
import org.springframework.http.HttpStatus;
import org.springframework.web.server.ResponseStatusException;
import java.util.*;
import static com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.*;
/** Persisted receiving authority. All mutations run inside the caller's batch-locked transaction. */
@Component
public class FbaReceivingStore {
private final FbaReceivingMapper mapper;
private final ObjectMapper json;
public FbaReceivingStore(FbaReceivingMapper mapper,ObjectMapper json) { this.mapper=mapper;this.json=json; }
public boolean hasPending(String batchKey,String customerId) { return mapper.pendingCount(batchKey,customerId)>0; }
/** Three bounded queries for calendar/work providers; never a per-batch query loop. */
public List<BatchReceiving> snapshots(List<String> batchKeys) {
if(batchKeys==null || batchKeys.isEmpty()) return List.of();
if(batchKeys.size()>200 || batchKeys.stream().anyMatch(k->k==null || !k.matches("FBA_APPOINTMENT:[1-9][0-9]*"))) throw new ResponseStatusException(HttpStatus.BAD_REQUEST,"每次最多读取200个有效批次");
var groups=new HashMap<String,List<Agreement>>();var pending=new HashMap<String,List<Pending>>();var plans=new HashMap<String,List<CustomerPlan>>();
for(var row:mapper.currentGroups(batchKeys)) groups.computeIfAbsent(row.batchKey(),k->new ArrayList<>()).add(read(row.snapshotJson(),Agreement.class));
for(var row:mapper.pendingGroups(batchKeys)) pending.computeIfAbsent(row.batchKey(),k->new ArrayList<>()).add(read(row.snapshotJson(),Pending.class));
for(var row:mapper.plans(batchKeys)) plans.computeIfAbsent(row.batchKey(),k->new ArrayList<>()).add(read(row.snapshotJson(),CustomerPlan.class));
return batchKeys.stream().distinct().map(k->new BatchReceiving(k,List.copyOf(groups.getOrDefault(k,List.of())),List.copyOf(pending.getOrDefault(k,List.of())),List.copyOf(plans.getOrDefault(k,List.of())))).toList();
}
public List<CustomerPlan> plans(String batchKey) { return mapper.plans(List.of(batchKey)).stream().map(r->read(r.snapshotJson(),CustomerPlan.class)).toList(); }
public CustomerPlan plan(String batchKey,String customerId) { return plans(batchKey).stream().filter(p->customerId.equals(p.customerId())).findFirst().orElse(null); }
public CustomerPlan savePlan(CustomerPlan input) {
var saved=new CustomerPlan(input.batchKey(),input.customerId(),normalized(input.details()),input.timezone());
mapper.deletePlan(saved.batchKey(),saved.customerId());
mapper.insertPlan(new FbaReceivingMapper.PlanRow(saved.batchKey(),saved.customerId(),write(saved)));return saved;
}
public void deletePlan(String batchKey,String customerId) { mapper.deletePlan(batchKey,customerId); }
/** Preserve each still-unconfirmed customer's old target before changing the aggregate header. */
public void freezeUnconfirmedPlans(FbaAppointmentResponse batch) {
if(!Set.of("PENDING","FAILED").contains(batch.appointmentStatus())) return;
var existing=plans(batch.batchKey()).stream().map(CustomerPlan::customerId).collect(java.util.stream.Collectors.toSet());
var covered=agreements(batch.batchKey()).stream().filter(a->"ACTIVE".equals(a.status())).flatMap(a->a.members().stream()).filter(m->batch.batchKey().equals(m.batchKey())).map(Member::customerId).collect(java.util.stream.Collectors.toSet());
var details=new FbaAppointmentDetails(batch.plannedDate(),batch.plannedAt(),batch.plannedEndAt(),null,null,null,batch.appointmentStatus(),batch.appointmentRemark(),null,null);
for(var customer:batch.items().stream().map(FbaAppointmentResponse.Item::customerId).distinct().toList())
if(!existing.contains(customer) && !covered.contains(customer)) savePlan(new CustomerPlan(batch.batchKey(),customer,details,batch.timezone()));
}
public Agreement currentAgreement(String batchKey,String customerId) {
var key=mapper.currentKey(batchKey,customerId);return key==null?null:agreement(key,false);
}
public List<Agreement> agreements(String batchKey) { return mapper.groupKeys(batchKey).stream().map(k->agreement(k,false)).toList(); }
public List<Pending> pending(String batchKey) { return mapper.pending(batchKey).stream().map(r->read(r.snapshotJson(),Pending.class)).toList(); }
public List<String> successorKeys(String batchKey) {
return mapper.appliedResults(batchKey).stream().map(r->read(r,Result.class)).flatMap(r->r.batches().stream())
.map(FbaAppointmentResponse::batchKey).filter(k->!k.equals(batchKey)).distinct().toList();
}
public List<com.baiyue.lastleg.appointmentplanning.AppointmentCoverageService.SourceRef> originalSources(String batchKey) {
return mapper.appliedChanges(batchKey).stream().map(r->read(r,Pending.class)).flatMap(p->p.originalMembers().stream())
.flatMap(m->m.sources().stream().map(s->new com.baiyue.lastleg.appointmentplanning.AppointmentCoverageService.SourceRef(m.customerId(),s.sourceKey(),s.destinationKey()))).distinct().toList();
}
public Agreement agreement(String key,boolean lock) {
var row=lock?mapper.lockGroup(key):mapper.group(key);
if(row==null) throw new ResponseStatusException(HttpStatus.NOT_FOUND,"收货预约不存在");
return read(row.snapshotJson(),Agreement.class);
}
public List<Member> members(FbaAppointmentResponse batch) {
return members(batch,null);
}
public Member member(FbaAppointmentResponse batch,String customerId) {return members(batch,customerId).stream().findFirst().orElse(null);}
private List<Member> members(FbaAppointmentResponse batch,String customerId) {
var labels=new HashMap<String,String>();for(var label:mapper.labels(batch.batchKey())) if(label.customerName()!=null) labels.putIfAbsent(label.customerId(),label.customerName());
var grouped=new LinkedHashMap<String,List<Source>>();
for(var i:batch.items()) {
if(customerId!=null && !customerId.equals(i.customerId())) continue;
if(i.packageCount()==null || i.packageCount()<=0) throw conflict("请先核实本批各票箱数,再调整收货范围");
grouped.computeIfAbsent(i.customerId(),ignored->new ArrayList<>()).add(new Source(i.sourceKey(),i.destinationKey(),i.packageCount(),i.sourceVersion(),i.cargoVerificationVersion(),i.inboundCode()));
}
return grouped.entrySet().stream().map(e->new Member(batch.batchKey(),batch.appointmentNo(),batch.version(),e.getKey(),List.copyOf(e.getValue()),labels.get(e.getKey()))).toList();
}
public Agreement create(List<Member> members,FbaAppointmentDetails details,String timezone,String evidence) {
details=normalized(details);
String destination=members.get(0).sources().get(0).destinationKey();
var group=new Agreement("RA-"+UUID.randomUUID().toString().replace("-","").toUpperCase(Locale.ROOT),0,"ACTIVE",destination,details,timezone,List.copyOf(members),evidence);
try {
mapper.insertGroup(row(group));mapper.revision(row(group));
for(var member:members) mapper.addMember(group.agreementKey(),member.batchKey(),member.customerId(),member.batchKey()+"/"+member.customerId());
} catch(org.springframework.dao.DuplicateKeyException duplicate) {throw conflict("该目的仓约号或客户覆盖已被其他有效预约使用");}
return group;
}
public Agreement update(Agreement old,List<Member> members,FbaAppointmentDetails details,String evidence) {
details=normalized(details);
if(Objects.equals(old.details(),details) && coverage(old.members()).equals(coverage(members))) return old;
var group=new Agreement(old.agreementKey(),old.version()+1,old.status(),old.destinationKey(),details,old.timezone(),List.copyOf(members),evidence);
try {mapper.updateGroup(row(group));mapper.revision(row(group));} catch(org.springframework.dao.DuplicateKeyException duplicate) {throw conflict("该目的仓约号已被其他有效预约使用");}return group;
}
public Agreement cancel(Agreement old,String status,String evidence) {
var group=new Agreement(old.agreementKey(),old.version()+1,status,old.destinationKey(),old.details(),old.timezone(),old.members(),evidence);
mapper.releaseGroup(old.agreementKey());mapper.updateGroup(row(group));mapper.revision(row(group));return group;
}
public void releaseMember(String group,String batch,String customer) { mapper.releaseMember(group,batch,customer); }
public void assertLegacyEditable(String batchKey) {
assertRepartitionable(batchKey);
if(!plans(batchKey).isEmpty()) throw conflict("本批已有按客户维护的计划,请按客户继续处理,不能整批覆盖或取消");
}
public void assertRepartitionable(String batchKey) {
if(!pending(batchKey).isEmpty()) throw conflict("本批有待仓库确认的变化,请从变化记录继续处理");
var active=agreements(batchKey).stream().filter(a->"ACTIVE".equals(a.status())).toList();
if(active.size()>1 || active.stream().anyMatch(a->a.members().stream().anyMatch(m->!m.batchKey().equals(batchKey))))
throw conflict("本批属于独立收货或跨批共同预约,请使用明确范围的收货预约操作");
if(!active.isEmpty()) {
var actual=new HashSet<>(mapper.liveSources(batchKey));
var covered=new HashSet<FbaReceivingMapper.LiveSource>();
for(var member:active.get(0).members()) {
if(!Objects.equals(mapper.currentKey(batchKey,member.customerId()),active.get(0).agreementKey())) throw conflict("本批客户覆盖关系已变化,请按当前收货范围操作");
for(var source:member.sources()) covered.add(new FbaReceivingMapper.LiveSource(member.customerId(),source.sourceKey(),source.destinationKey(),source.packageCount()));
}
if(!actual.equals(covered)) throw conflict("本批仅部分客户已取得确认,请分别处理当前客户或明确共同预约范围");
}
}
/** Existing single-batch commands produce the same receiving authority; no read-time reconstruction. */
public void synchronize(Object result) {
if(!(result instanceof FbaAppointmentResponse batch)) return;
var active=agreements(batch.batchKey()).stream().filter(a->"ACTIVE".equals(a.status())).toList();
if("SUCCESS".equals(batch.appointmentStatus())) {
var details=new FbaAppointmentDetails(batch.plannedDate(),batch.plannedAt(),batch.plannedEndAt(),batch.confirmationNo(),batch.appointmentAt(),batch.appointmentEndAt(),"SUCCESS",batch.appointmentRemark(),null,null);
if(active.isEmpty()) create(members(batch),details,batch.timezone(),"登记收货方正式确认");
else update(active.get(0),members(batch),details,"登记收货方正式确认");
} else for(var agreement:active) cancel(agreement,"CANCELLED".equals(batch.appointmentStatus())?"CANCELLED":"SUPERSEDED","原预约结果已变更");
}
public Pending change(String key,boolean lock) {
var row=lock?mapper.lockChange(key):mapper.change(key);
if(row==null) throw new ResponseStatusException(HttpStatus.NOT_FOUND,"预约变化记录不存在");
return read(row.snapshotJson(),Pending.class);
}
public void save(Pending change) { mapper.insertChange(new FbaReceivingMapper.ChangeRow(change.changeKey(),change.version(),change.status(),change.intent().batchKey(),change.intent().customerId(),write(change),null)); }
public void finish(Pending old,String status,Result result) {
var updated=new Pending(old.changeKey(),old.version()+1,status,old.intent(),old.originalMembers(),old.originalAgreements(),old.followup());
mapper.finishChange(new FbaReceivingMapper.ChangeRow(old.changeKey(),updated.version(),status,old.intent().batchKey(),old.intent().customerId(),write(updated),write(result)));
}
private static List<String> coverage(List<Member> members) {
return members.stream().flatMap(m->m.sources().stream().map(s->List.of(m.batchKey(),m.customerId(),s.sourceKey(),s.destinationKey(),s.packageCount(),Objects.toString(s.sourceVersion(),""),Objects.toString(s.verificationVersion(),"")).toString())).sorted().toList();
}
static FbaAppointmentDetails normalized(FbaAppointmentDetails d) {
return new FbaAppointmentDetails(d.plannedAt()==null?d.plannedDate():d.plannedAt().toLocalDate(),d.plannedAt(),d.plannedEndAt(),d.confirmationNo()==null?null:d.confirmationNo().trim(),d.appointmentAt(),d.appointmentEndAt(),d.appointmentStatus(),d.appointmentRemark()==null?null:d.appointmentRemark().trim(),d.startOffset(),d.endOffset());
}
private FbaReceivingMapper.GroupRow row(Agreement a) { return new FbaReceivingMapper.GroupRow(a.agreementKey(),a.version(),a.status(),a.destinationKey(),"ACTIVE".equals(a.status())?a.details().confirmationNo():null,write(a)); }
public String write(Object value) {try{return json.writeValueAsString(value);}catch(Exception e){throw new IllegalStateException("预约记录无法保存",e);} }
public <T>T read(String value,Class<T> type) {try{return json.readValue(value,type);}catch(Exception e){throw new IllegalStateException("预约记录无法读取",e);} }
static ResponseStatusException conflict(String message) { return new ResponseStatusException(HttpStatus.CONFLICT,message); }
}
@@ -31,6 +31,8 @@ public class InboundFbaAppointmentService {
@Resource private AppointmentBatchPreviewMapper flags;
@Resource private AppointmentCargoVerificationService verification;
@Resource private ObjectMapper json;
@Resource private com.baiyue.lastleg.inboundplan.receiving.FbaReceivingStore receiving;
@Resource private com.baiyue.lastleg.inboundplan.receiving.FbaAppointmentWorkPort receivingWork;
@Resource private com.baiyue.lastleg.inboundplan.work.AppointmentWorkAppointmentChanges workChanges;
@Transactional(readOnly=true)
@@ -38,7 +40,7 @@ public class InboundFbaAppointmentService {
if(from==null || to==null || !to.isAfter(from) || ChronoUnit.DAYS.between(from,to)>42) throw bad("日历查询范围必须为1至42天");
var query=new LambdaQueryWrapper<InboundFbaAppointment>()
.isNull(InboundFbaAppointment::getReplacementNosJson)
.notIn(InboundFbaAppointment::getAppointmentStatus,List.of("CANCELLED","FAILED"))
.ne(InboundFbaAppointment::getAppointmentStatus,"CANCELLED")
.apply("COALESCE(appointment_at,planned_at,planned_date)<{1} AND (COALESCE(appointment_at,planned_at,planned_date)>={0} OR CASE WHEN appointment_at IS NOT NULL THEN appointment_end_at WHEN planned_at IS NOT NULL THEN planned_end_at ELSE NULL END>{0})",from.atStartOfDay(),to.atStartOfDay())
.orderByAsc(InboundFbaAppointment::getId);
return responses(appointmentMapper.selectList(query));
@@ -49,8 +51,7 @@ public class InboundFbaAppointmentService {
var query=new LambdaQueryWrapper<InboundFbaAppointment>()
.isNull(InboundFbaAppointment::getReplacementNosJson)
.isNull(InboundFbaAppointment::getAppointmentAt).isNull(InboundFbaAppointment::getPlannedAt)
.isNull(InboundFbaAppointment::getPlannedDate)
.notIn(InboundFbaAppointment::getAppointmentStatus,List.of("CANCELLED","FAILED"));
.isNull(InboundFbaAppointment::getPlannedDate);
long total=appointmentMapper.selectCount(query);
var rows=appointmentMapper.selectList(query.orderByDesc(InboundFbaAppointment::getCreatedAt)
.last("LIMIT "+size+" OFFSET "+(long)(page-1)*size));
@@ -116,12 +117,17 @@ public class InboundFbaAppointmentService {
try {id=Long.parseLong(key.substring("FBA_APPOINTMENT:".length()));if(id<=0 || !key.equals(batchKey(id))) throw new IllegalArgumentException();}
catch(RuntimeException error) {throw bad("批次关联键无效");}
var appointment=appointmentMapper.selectById(id);if(appointment==null) throw missing();
var rows=items(id);var snapshot=response(appointment,rows);
var rows=items(id);if(appointment.getReplacementNosJson()==null) rows=rows.stream().filter(i->Boolean.TRUE.equals(i.getActive())).toList();var snapshot=response(appointment,rows);
var originalRefs=originalSourceRefs(snapshot);
var reasons=new ArrayList<String>();var actions=new ArrayList<String>();
if(appointment.getReplacementNosJson()!=null) reasons.add("原批次已被分批或合并,请进入替代批次");
if(!"PENDING".equals(appointment.getAppointmentStatus())) reasons.add("仅未取得正式确认的活动批次可分批或合并");
if(queryMapper.countActiveDispatches(id)>0) reasons.add("已关联排车,请先处理运输安排");
if(!reasons.isEmpty()) return new FbaAppointmentBatchContext(snapshot,List.of(),List.of(),List.copyOf(reasons));
try {receiving.assertRepartitionable(key);} catch(ResponseStatusException error) {
if(error.getStatusCode().value()!=409) throw error;
reasons.add(error.getReason());
}
if(!reasons.isEmpty()) return new FbaAppointmentBatchContext(snapshot,List.of(),List.of(),List.copyOf(reasons),currentBatches(snapshot,originalRefs),originalRefs);
var refs=rows.stream().map(item->new AppointmentCoverageService.SourceRef(item.getCustomerId(),item.getSourceKey(),"FBA:"+appointment.getFbaWarehouseCode())).toList();
var resolved=sources.resolveBatch(refs);var evidence=verification(resolved);
if(resolved.stream().anyMatch(s->!Objects.equals(s.originWarehouseId(),appointment.getOriginWarehouseId()) || !Objects.equals(s.destinationWarehouseCode(),appointment.getFbaWarehouseCode()))) reasons.add("原来源起讫点已变化,请先复核原批次");
@@ -143,7 +149,42 @@ public class InboundFbaAppointmentService {
if(allocations.stream().mapToLong(a->a.packageCount()).sum()>1) actions.add("SPLIT");
actions.add("MERGE");
}
return new FbaAppointmentBatchContext(snapshot,List.copyOf(allocations),List.copyOf(actions),reasons.stream().distinct().toList());
return new FbaAppointmentBatchContext(snapshot,List.copyOf(allocations),List.copyOf(actions),reasons.stream().distinct().toList(),currentBatches(snapshot,originalRefs),originalRefs);
}
private List<FbaAppointmentBatchContext.SourceRef> originalSourceRefs(FbaAppointmentResponse original) {
var refs=new LinkedHashSet<FbaAppointmentBatchContext.SourceRef>();
original.items().forEach(item->refs.add(new FbaAppointmentBatchContext.SourceRef(item.customerId(),item.sourceKey(),item.destinationKey())));
receiving.originalSources(original.batchKey()).forEach(ref->refs.add(new FbaAppointmentBatchContext.SourceRef(ref.customerId(),ref.sourceKey(),ref.destinationKey())));
return List.copyOf(refs);
}
/** Follow all replacement branches. Shared merge descendants are returned once, never just the first branch. */
private List<FbaAppointmentResponse> currentBatches(FbaAppointmentResponse original,List<FbaAppointmentBatchContext.SourceRef> originalRefs) {
var sourceRefs=new HashSet<>(originalRefs);
var graph=new LinkedHashMap<String,FbaAppointmentResponse>();graph.put(original.batchKey(),original);
var pending=new LinkedHashSet<>(original.replacementBatchKeys());
pending.addAll(receiving.successorKeys(original.batchKey()));
while(!pending.isEmpty()) {
if(graph.size()+pending.size()>1000) throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"批次去向超过读取范围,请缩小范围后核对");
var keys=new ArrayList<>(pending);pending.clear();
for(int start=0;start<keys.size();start+=200) {
for(var batch:batchSnapshots(keys.subList(start,Math.min(start+200,keys.size())))) graph.put(batch.batchKey(),batch);
}
for(var batch:graph.values()) {
var nextKeys=new ArrayList<>(batch.replacementBatchKeys());nextKeys.addAll(receiving.successorKeys(batch.batchKey()));
for(var next:nextKeys) if(!graph.containsKey(next)) pending.add(next);
}
}
var visiting=new HashSet<String>();var visited=new HashSet<String>();
verifyReplacementGraph(original.batchKey(),graph,visiting,visited);
return graph.values().stream().filter(batch->batch.replacementAppointmentNos().isEmpty())
.filter(batch->batch.items().stream().anyMatch(item->sourceRefs.contains(new FbaAppointmentBatchContext.SourceRef(item.customerId(),item.sourceKey(),item.destinationKey()))))
.toList();
}
private static void verifyReplacementGraph(String key,Map<String,FbaAppointmentResponse> graph,Set<String> visiting,Set<String> visited) {
if(visited.contains(key)) return;
if(!visiting.add(key)) throw conflict("批次去向存在循环,请核对原业务记录");
for(var next:graph.get(key).replacementBatchKeys()) verifyReplacementGraph(next,graph,visiting,visited);
visiting.remove(key);visited.add(key);
}
/** Original persisted appointment facts only; reading work status must not run split/merge eligibility. */
@Transactional(readOnly=true)
@@ -165,15 +206,18 @@ public class InboundFbaAppointmentService {
return command("UPDATE:"+appointmentNo,operatorId,request.idempotencyKey(),request,()->{
var appointment=find(appointmentNo,true);if(appointment==null) throw missing();
requireCurrent(appointment);
receiving.assertLegacyEditable(batchKey(appointment.getId()));
if(!Objects.equals(appointment.getVersion(),request.version())) throw conflict("预约已更新,请重新读取后操作");
if(queryMapper.countActiveDispatches(appointment.getId())>0) throw conflict("已关联有效排车,请先核对并处理运输安排后再改约");
var current=items(appointment.getId());
var current=items(appointment.getId()).stream().filter(i->Boolean.TRUE.equals(i.getActive())).toList();
var before=response(appointment,current);
String status=status(request.details().appointmentStatus());
if("SUCCESS".equals(appointment.getAppointmentStatus()) && !"SUCCESS".equals(status)) throw bad("已确认预约不能直接退回待确认或未约成,请使用取消操作并记录原因");
if("FAILED".equals(status)) {
if(clean(request.details().appointmentRemark())==null) throw bad("未约成时请填写原因");
// Keep the last planned and confirmed times as historical facts.
// Failure releases receiving coverage, not the cargo allocation owned by this batch.
appointment.setAppointmentStatus(status);appointment.setAppointmentRemark(request.details().appointmentRemark().trim());
current.forEach(item->{item.setActive(false);itemMapper.updateById(item);});
clearConfirmation(appointment);
} else {
var refs=current.stream().map(item->new AppointmentCoverageService.SourceRef(item.getCustomerId(),item.getSourceKey(),"FBA:"+appointment.getFbaWarehouseCode())).toList();
var resolved=resolveForWrite(refs);
@@ -183,6 +227,7 @@ public class InboundFbaAppointmentService {
apply(appointment,request.details(),resolved,evidence);
writeItems(appointment,resolved,current,evidence,Map.of());
}
if(before.equals(response(appointment,items(appointment.getId())))) return before;
appointment.setVersion(request.version()+1);appointment.setUpdatedAt(now());
if(appointmentMapper.update(appointment,new LambdaQueryWrapper<InboundFbaAppointment>().eq(InboundFbaAppointment::getId,appointment.getId()).eq(InboundFbaAppointment::getVersion,request.version()))!=1) throw conflict("预约版本已变化");
return response(appointment,items(appointment.getId()));
@@ -196,15 +241,26 @@ public class InboundFbaAppointmentService {
command("CANCEL:"+appointmentNo,operatorId,"CANCEL:"+appointmentNo+":"+version,cancellation,()->{
var appointment=find(appointmentNo,true);if(appointment==null) throw missing();
requireCurrent(appointment);
receiving.assertLegacyEditable(batchKey(appointment.getId()));
if(!Objects.equals(appointment.getVersion(),version)) throw conflict("预约已更新,请重新读取后操作");
if(queryMapper.countActiveDispatches(appointment.getId())>0) throw conflict("预约货物已关联排车,不允许取消,请先处理关联排车");
appointment.setAppointmentStatus("CANCELLED");
appointment.setVersion(version+1);appointment.setUpdatedAt(now());appointmentMapper.updateById(appointment);
var rows=items(appointment.getId());rows.forEach(item->{item.setActive(false);itemMapper.updateById(item);});
return response(appointment,rows);
appointment.setVersion(version+1);appointment.setUpdatedAt(now());
var rows=items(appointment.getId());
// Persist the cancelled plan in the immutable command history; the active demand has no valid date/confirmation.
var cancelledSnapshot=response(appointment,rows);
clearConfirmation(appointment);
appointment.setPlannedDate(null);appointment.setPlannedAt(null);appointment.setPlannedEndAt(null);
appointmentMapper.updateById(appointment);
return cancelledSnapshot;
});
}
private static void clearConfirmation(InboundFbaAppointment appointment) {
appointment.setConfirmationNo(null);appointment.setAppointmentAt(null);appointment.setAppointmentEndAt(null);
appointment.setAppointmentStartUtc(null);appointment.setAppointmentEndUtc(null);
}
@Transactional(isolation=org.springframework.transaction.annotation.Isolation.READ_COMMITTED)
public RepartitionFbaAppointmentsResponse repartition(String operator,RepartitionFbaAppointmentsRequest request) {
validateRepartition(request);
@@ -213,7 +269,11 @@ public class InboundFbaAppointmentService {
var originals=new ArrayList<InboundFbaAppointment>();
var originalItems=new LinkedHashMap<Long,List<InboundFbaAppointmentItem>>();
var refs=new LinkedHashMap<String,AppointmentCoverageService.SourceRef>();
for(var intent:request.originals().stream().sorted(Comparator.comparing(RepartitionFbaAppointmentsRequest.Original::appointmentNo)).toList()) {
var intents=request.originals().stream().collect(Collectors.toMap(RepartitionFbaAppointmentsRequest.Original::appointmentNo,i->i));
var ordered=request.originals().stream().map(i->{var row=find(i.appointmentNo(),false);if(row==null)throw missing();return row;})
.sorted(Comparator.comparing(row->batchKey(row.getId()))).toList();
for(var candidate:ordered) {
var intent=intents.get(candidate.getAppointmentNo());
var row=find(intent.appointmentNo(),true);if(row==null) throw missing();
requireCurrent(row);
if(!Objects.equals(row.getVersion(),intent.version())) throw conflict("原预约已更新,请重新核对分批资料");
@@ -222,7 +282,8 @@ public class InboundFbaAppointmentService {
if(queryMapper.countActiveDispatches(row.getId())>0) throw conflict("原批次已有关联排车,请先处理运输安排");
if(!originals.isEmpty() && (!Objects.equals(originals.get(0).getOriginWarehouseId(),row.getOriginWarehouseId())
|| !Objects.equals(originals.get(0).getFbaWarehouseCode(),row.getFbaWarehouseCode()))) throw bad("仅同一起运仓、同一目的仓可以合并");
var items=items(row.getId());
receiving.assertRepartitionable(batchKey(row.getId()));
var items=items(row.getId()).stream().filter(i->Boolean.TRUE.equals(i.getActive())).toList();
if(items.isEmpty() || items.stream().anyMatch(item->!Boolean.TRUE.equals(item.getActive()))) throw conflict("原批次货物已变化,请重新核对");
for(var item:items) {
var ref=new AppointmentCoverageService.SourceRef(item.getCustomerId(),item.getSourceKey(),"FBA:"+row.getFbaWarehouseCode());
@@ -284,6 +345,9 @@ public class InboundFbaAppointmentService {
row.setReplacementNosJson(replacementsJson);row.setReplacementKeysJson(replacementKeysJson);row.setRepartitionReason(request.reason().trim());row.setVersion(row.getVersion()+1);row.setUpdatedAt(now());
if(appointmentMapper.updateById(row)!=1) throw conflict("原批次替代关系未保存");
}
var oldMembers=originalItems.entrySet().stream().flatMap(e->e.getValue().stream().map(i->new com.baiyue.lastleg.inboundplan.receiving.FbaAppointmentWorkPort.Member(batchKey(e.getKey()),i.getCustomerId()))).distinct().toList();
var newMembers=replacements.stream().flatMap(b->b.items().stream().map(i->new com.baiyue.lastleg.inboundplan.receiving.FbaAppointmentWorkPort.Member(b.batchKey(),i.customerId()))).distinct().toList();
receivingWork.inheritFollowup(oldMembers,newMembers,operator);
return new RepartitionFbaAppointmentsResponse(originals.stream().map(row->batchKey(row.getId())).toList(),originals.stream().map(InboundFbaAppointment::getAppointmentNo).toList(),List.copyOf(replacements));
});
}
@@ -304,6 +368,164 @@ public class InboundFbaAppointmentService {
}
}
/** Receiving orchestration acquires transport locks first, then invokes this boundary. */
@Transactional(propagation=org.springframework.transaction.annotation.Propagation.MANDATORY)
public FbaAppointmentResponse lockReceivingBatch(String key) {
long id=parseBatchKey(key);
var row=appointmentMapper.selectOne(new LambdaQueryWrapper<InboundFbaAppointment>().eq(InboundFbaAppointment::getId,id).last("FOR UPDATE"));
if(row==null) throw missing();requireCurrent(row);
return response(row,items(id));
}
/** Explicit adoption updates only the chosen customer's evidence, never their allocated amount. */
@Transactional(propagation=org.springframework.transaction.annotation.Propagation.MANDATORY)
public FbaAppointmentResponse refreshReceivingCargo(String key,String customerId,List<com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.CargoVersion> expected) {
var snapshot=lockReceivingBatch(key);var row=appointmentMapper.selectById(snapshot.id());
var current=items(row.getId()).stream().filter(i->Boolean.TRUE.equals(i.getActive())).toList();
var selected=current.stream().filter(i->Objects.equals(customerId,i.getCustomerId())).toList();
if(selected.isEmpty() || expected==null || expected.size()!=selected.size() || expected.size()>100) throw bad("请完整核对本客户当前批次的每票来源资料");
var versions=new HashMap<String,com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.CargoVersion>();
for(var value:expected) {
if(value==null || clean(value.sourceKey())==null || clean(value.destinationKey())==null || clean(value.expectedSourceVersion())==null || value.expectedVerificationVersion()!=null && value.expectedVerificationVersion()<1 || versions.putIfAbsent(value.sourceKey(),value)!=null)
throw bad("来源不能重复,必须明确当前来源及核实版本");
}
var refs=selected.stream().map(i->new AppointmentCoverageService.SourceRef(i.getCustomerId(),i.getSourceKey(),"FBA:"+i.getFbaWarehouseCode())).toList();
var resolved=resolveForWrite(refs);var evidence=verification(resolved);
var bySource=resolved.stream().collect(Collectors.toMap(AppointmentSourceService.Source::sourceKey,s->s));
for(var item:selected) {
var requested=versions.get(item.getSourceKey());var source=bySource.get(item.getSourceKey());
var verified=source==null?null:evidence.get(source.sourceKey());
if(requested==null || source==null || !Objects.equals(customerId,source.customerId()) || !Objects.equals("FBA:"+item.getFbaWarehouseCode(),source.destinationKey())
|| !Objects.equals(source.destinationKey(),requested.destinationKey()) || !Objects.equals(row.getOriginWarehouseId(),source.originWarehouseId())
|| !Objects.equals(source.sourceVersion(),requested.expectedSourceVersion()) || !Objects.equals(verified==null?null:verified.version(),requested.expectedVerificationVersion()))
throw conflict("本票归属、目的地或最新资料版本已变化,请重新预览后确认");
var total=quantity(source,evidence);var count=item.getPackageCount();
var other=queryMapper.allocation(customerId,source.sourceKey(),source.orderId(),row.getId());
if(count==null || count<=0 || total==null || other.unknownCount()>0 || (long)count+other.packages()>total) throw conflict("最新核实量不足以覆盖当前全部批次,请先处理数量差异,不能直接采用资料");
item.setSourceVersion(source.sourceVersion());item.setCargoVerificationVersion(verified==null?null:verified.version());
try {item.setShipmentReferencesJson(verified==null?null:json.writeValueAsString(verified.shipmentReferences()));}
catch(com.fasterxml.jackson.core.JsonProcessingException error) {throw new IllegalStateException("核实货件资料无法保存",error);}
}
if(snapshot.equals(response(row,current))) return snapshot;
selected.forEach(itemMapper::updateById);
row.setVersion(row.getVersion()+1);row.setUpdatedAt(now());appointmentMapper.updateById(row);
var result=response(row,current);workChanges.changed(result);return result;
}
@Transactional(propagation=org.springframework.transaction.annotation.Propagation.MANDATORY)
public String validateReceiving(List<com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.Member> members,FbaAppointmentDetails details) {
if(members==null || members.isEmpty() || members.size()>100 || details==null || !"SUCCESS".equals(details.appointmentStatus())) throw bad("请填写明确覆盖货物的正式收货确认");
return validateReceivingDetails(members,details);
}
@Transactional(propagation=org.springframework.transaction.annotation.Propagation.MANDATORY)
public String validateReceivingPlan(com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.Member member,FbaAppointmentDetails details) {
if(details==null || !Set.of("PENDING","FAILED").contains(details.appointmentStatus())) throw bad("客户计划只允许待确认或未约成,正式确认请使用登记确认操作");
if(details.confirmationNo()!=null || details.appointmentAt()!=null || details.appointmentEndAt()!=null) throw bad("客户计划不能填写外部约号或确认时段");
if(details.plannedAt()==null && (clean(details.startOffset())!=null || clean(details.endOffset())!=null)) throw bad("未填写计划开始时刻,不能指定计划时差");
if("FAILED".equals(details.appointmentStatus()) && clean(details.appointmentRemark())==null) throw bad("未约成时请填写原因");
var timezone=validateReceivingDetails(List.of(member),details);
if(details.plannedDate()!=null || details.plannedAt()!=null) {
if(timezone==null || !ZoneId.getAvailableZoneIds().contains(timezone)) throw bad("填写目标日期前请补齐目的仓IANA时区");
try {ZoneId.of(timezone);} catch(DateTimeException error) {throw bad("目的仓IANA时区无效");}
}
if(details.plannedAt()!=null) {
try {AppointmentTimePolicy.resolve(details.plannedAt(),details.plannedEndAt(),timezone,offset(details.startOffset()),offset(details.endOffset()));}
catch(IllegalArgumentException | DateTimeException error) {throw bad(error.getMessage());}
}
return timezone;
}
private String validateReceivingDetails(List<com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.Member> members,FbaAppointmentDetails details) {
var refs=members.stream().flatMap(m->m.sources().stream().map(s->new AppointmentCoverageService.SourceRef(m.customerId(),s.sourceKey(),s.destinationKey()))).distinct().toList();
var resolved=resolveForWrite(refs);var evidence=verification(resolved);var bySource=resolved.stream().collect(Collectors.toMap(AppointmentSourceService.Source::sourceKey,s->s));
for(var m:members) for(var s:m.sources()) {
var current=bySource.get(s.sourceKey());var verified=evidence.get(s.sourceKey());
if(current==null || !Objects.equals(current.customerId(),m.customerId()) || !Objects.equals(current.destinationKey(),s.destinationKey())
|| !Objects.equals(current.sourceVersion(),s.sourceVersion()) || !Objects.equals(verified==null?null:verified.version(),s.verificationVersion())) throw conflict("原来源或核实版本已变化,请先核对货物");
}
var grouped=group(resolved);var row=new InboundFbaAppointment();row.setOriginWarehouseId(grouped.originWarehouseId());row.setFbaWarehouseCode(resolved.get(0).destinationWarehouseCode());
apply(row,details,resolved,evidence);return row.getTimezone();
}
@Transactional(propagation=org.springframework.transaction.annotation.Propagation.MANDATORY)
public List<FbaAppointmentResponse> moveReceivingCustomer(String operator,com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.ChangeIntent intent) {
var original=find(lockReceivingBatch(intent.batchKey()).appointmentNo(),false);
if(!Objects.equals(original.getVersion(),intent.batchVersion())) throw conflict("原批次版本已变化,请重新核对调整范围");
var all=items(original.getId()).stream().filter(i->Boolean.TRUE.equals(i.getActive())).toList();
receiving.freezeUnconfirmedPlans(response(original,all));
var selected=all.stream().filter(i->intent.customerId().equals(i.getCustomerId())).toList();
if(selected.isEmpty() || selected.size()!=intent.quantities().size()) throw bad("请完整选择本客户的各票货物");
var amounts=new HashMap<String,com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.Quantity>();
for(var q:intent.quantities()) if(amounts.put(q.sourceKey(),q)!=null) throw bad("逐票来源不能重复");
var refs=selected.stream().map(i->new AppointmentCoverageService.SourceRef(i.getCustomerId(),i.getSourceKey(),"FBA:"+i.getFbaWarehouseCode())).toList();
var resolved=resolveForWrite(refs);var evidence=verification(resolved);var bySource=resolved.stream().collect(Collectors.toMap(AppointmentSourceService.Source::sourceKey,s->s));
var laterAmounts=new LinkedHashMap<String,Integer>();
for(var item:selected) {
var q=amounts.get(item.getSourceKey());var src=bySource.get(item.getSourceKey());var v=evidence.get(item.getSourceKey());
if(q==null || src==null || !Objects.equals(src.destinationKey(),q.destinationKey()) || !Objects.equals(src.sourceVersion(),q.expectedSourceVersion())
|| !Objects.equals(v==null?null:v.version(),q.expectedVerificationVersion()) || !Objects.equals(src.sourceVersion(),item.getSourceVersion())) throw conflict("原来源或核实资料版本已变化");
Integer total=quantity(src,evidence),count=item.getPackageCount();
var other=queryMapper.allocation(src.customerId(),src.sourceKey(),src.orderId(),original.getId());
if(count==null || total==null || other.unknownCount()>0 || (long)count+other.packages()>total || q.retainedPackages()<0 || q.retainedPackages()>count) throw conflict("本次保留量及原来源分配总量不一致");
int later=count-q.retainedPackages();if(later>0) laterAmounts.put(item.getSourceKey(),later);
item.setPackageCount(q.retainedPackages());item.setActive(q.retainedPackages()>0);itemMapper.updateById(item);
}
if(laterAmounts.isEmpty()) throw bad("本次货量没有变化,无需创建变化记录");
var laterRows=resolved.stream().filter(s->laterAmounts.containsKey(s.sourceKey())).toList();
var later=new InboundFbaAppointment();later.setId(IdWorker.getId());later.setAppointmentNo("AP-"+LocalDate.now(ZoneOffset.UTC).toString().replace("-","")+"-"+UUID.randomUUID().toString().replace("-","").substring(0,12).toUpperCase(Locale.ROOT));
later.setOriginWarehouseId(original.getOriginWarehouseId());later.setFbaWarehouseCode(original.getFbaWarehouseCode());later.setVersion(0);later.setSelectedCount(laterRows.size());later.setCreatedBy(operator);later.setCreatedAt(now());later.setUpdatedAt(now());
apply(later,new FbaAppointmentDetails(intent.laterDate(),null,null,null,null,null,"PENDING",intent.reason(),null,null),laterRows,evidence);
appointmentMapper.insert(later);writeItems(later,laterRows,List.of(),evidence,laterAmounts);
var retained=all.stream().filter(i->Boolean.TRUE.equals(i.getActive())).toList();
if(retained.stream().noneMatch(i->intent.customerId().equals(i.getCustomerId()))) receiving.deletePlan(intent.batchKey(),intent.customerId());
original.setSelectedCount(retained.size());original.setVersion(original.getVersion()+1);original.setUpdatedAt(now());
if(retained.isEmpty()) {original.setReplacementNosJson(receiving.write(List.of(later.getAppointmentNo())));original.setReplacementKeysJson(receiving.write(List.of(batchKey(later.getId()))));original.setRepartitionReason(intent.reason());}
appointmentMapper.updateById(original);
return List.of(response(original,retained),response(later,items(later.getId())));
}
/** Header is only an aggregate projection; per-customer receiving identity remains authoritative. */
@Transactional(propagation=org.springframework.transaction.annotation.Propagation.MANDATORY)
public FbaAppointmentResponse projectReceiving(String key,boolean cancelled) {
var row=appointmentMapper.selectById(parseBatchKey(key));var rows=items(row.getId()).stream().filter(i->Boolean.TRUE.equals(i.getActive())).toList();
var groups=rows.stream().map(InboundFbaAppointmentItem::getCustomerId).distinct().map(c->receiving.currentAgreement(key,c)).toList();
var active=groups.stream().filter(Objects::nonNull).toList();
clearConfirmation(row);
if(!groups.isEmpty() && groups.size()==active.size() && active.stream().map(a->a.agreementKey()).distinct().count()==1) {
var a=active.get(0);var d=a.details();row.setAppointmentStatus("SUCCESS");row.setConfirmationNo(d.confirmationNo());row.setAppointmentAt(d.appointmentAt());row.setAppointmentEndAt(d.appointmentEndAt());row.setTimezone(a.timezone());
var window=AppointmentTimePolicy.resolve(d.appointmentAt(),d.appointmentEndAt(),a.timezone(),offset(d.startOffset()),offset(d.endOffset()));
row.setAppointmentStartUtc(LocalDateTime.ofInstant(window.start(),ZoneOffset.UTC));row.setAppointmentEndUtc(window.end()==null?null:LocalDateTime.ofInstant(window.end(),ZoneOffset.UTC));
row.setPlannedDate(d.plannedDate());row.setPlannedAt(d.plannedAt());row.setPlannedEndAt(d.plannedEndAt());
} else {
var plans=receiving.plans(key).stream().collect(Collectors.toMap(com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.CustomerPlan::customerId,p->p));
var customerIds=rows.stream().map(InboundFbaAppointmentItem::getCustomerId).distinct().toList();
var details=new ArrayList<FbaAppointmentDetails>();var zones=new ArrayList<String>();
for(int i=0;i<customerIds.size();i++) {
var agreement=groups.get(i);var plan=plans.get(customerIds.get(i));
if(agreement!=null) {details.add(agreement.details());zones.add(agreement.timezone());}
else if(plan!=null) {details.add(plan.details());zones.add(plan.timezone());}
}
boolean complete=details.size()==customerIds.size() && !details.isEmpty();
boolean failed=complete && active.isEmpty() && details.stream().allMatch(d->"FAILED".equals(d.appointmentStatus()));
row.setAppointmentStatus(failed?"FAILED":cancelled && active.isEmpty() && plans.isEmpty()?"CANCELLED":"PENDING");
row.setPlannedDate(null);row.setPlannedAt(null);row.setPlannedEndAt(null);row.setAppointmentRemark(null);
if(complete) {
var first=details.get(0);var zone=zones.get(0);
if(details.stream().allMatch(d->Objects.equals(d.plannedDate(),first.plannedDate()) && Objects.equals(d.plannedAt(),first.plannedAt()) && Objects.equals(d.plannedEndAt(),first.plannedEndAt())) && zones.stream().allMatch(z->Objects.equals(z,zone))) {
row.setPlannedDate(first.plannedDate());row.setPlannedAt(first.plannedAt());row.setPlannedEndAt(first.plannedEndAt());row.setTimezone(zone);
}
if(details.stream().allMatch(first::equals)) row.setAppointmentRemark(first.appointmentRemark());
}
}
row.setVersion(row.getVersion()+1);row.setUpdatedAt(now());appointmentMapper.updateById(row);
var snapshot=response(row,rows);workChanges.changed(snapshot);return snapshot;
}
@Transactional(propagation=org.springframework.transaction.annotation.Propagation.MANDATORY)
public void receivingChanged(List<FbaAppointmentResponse> snapshots) {snapshots.forEach(workChanges::changed);}
private static long parseBatchKey(String key) {
try {long id=Long.parseLong(key.substring("FBA_APPOINTMENT:".length()));if(id>0 && key.equals(batchKey(id)))return id;} catch(RuntimeException ignored) { }
throw bad("批次关联键无效");
}
private List<AppointmentSourceService.Source> resolveForWrite(List<AppointmentCoverageService.SourceRef> refs) {
sources.lockForWrite(refs);
var resolved=sources.resolveBatch(refs);
@@ -405,6 +627,7 @@ public class InboundFbaAppointmentService {
try {commands.begin(key,hash,operation,operator,now());}
catch(DuplicateKeyException duplicate){return replay(commands.find(key),hash,type);}
var response=action.get();
receiving.synchronize(response);
workChanges.changed(response);
try {if(commands.finish(key,appointmentNumber.apply(response),commandSnapshot(response,request))!=1)throw conflict("预约命令结果未能保存");}
catch(com.fasterxml.jackson.core.JsonProcessingException error){throw new IllegalStateException("无法保存预约命令结果",error);}
@@ -460,6 +683,7 @@ public class InboundFbaAppointmentService {
return rows.stream().map(row->response(row,grouped.getOrDefault(row.getId(),List.of()))).toList();
}
private FbaAppointmentResponse response(InboundFbaAppointment row,List<InboundFbaAppointmentItem> items) {
if(row.getReplacementNosJson()==null) items=items.stream().filter(i->Boolean.TRUE.equals(i.getActive())).toList();
return new FbaAppointmentResponse(row.getId(),row.getAppointmentNo(),row.getConfirmationNo(),row.getPlannedDate(),row.getPlannedAt(),row.getPlannedEndAt(),row.getAppointmentAt(),row.getAppointmentEndAt(),
utc(row.getAppointmentStartUtc()),utc(row.getAppointmentEndUtc()),row.getTimezone(),row.getAppointmentStatus(),row.getAppointmentRemark(),row.getOriginWarehouseId(),row.getFbaWarehouseCode(),row.getSelectedCount(),row.getCreatedBy(),row.getCreatedAt(),row.getUpdatedAt(),
items.stream().map(item->new FbaAppointmentResponse.Item(item.getId(),item.getSourceKey(),item.getSourceVersion(),item.getOrderId(),item.getInboundCode(),item.getFbaWarehouseCode(),item.getCustomerId(),item.getPackageCount(),item.getCargoVerificationVersion(),shipmentReferences(item),"FBA:"+item.getFbaWarehouseCode())).toList(),row.getVersion(),stringList(row.getReplacementNosJson()),row.getRepartitionReason(),batchKey(row.getId()),stringList(row.getReplacementKeysJson()));
@@ -4,6 +4,8 @@ import com.baiyue.lastleg.appointmentwork.*;
import com.baiyue.lastleg.inboundplan.dto.FbaAppointmentResponse;
import com.baiyue.lastleg.inboundplan.dto.RepartitionFbaAppointmentsResponse;
import org.springframework.stereotype.Component;
import com.baiyue.lastleg.inboundplan.receiving.FbaReceivingStore;
import com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.*;
import java.time.*;
import java.util.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
@@ -12,7 +14,8 @@ import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
public class AppointmentWorkAppointmentChanges {
private final AppointmentWorkCommitmentService commitments;
private final AppointmentWorkChanges changes;
public AppointmentWorkAppointmentChanges(AppointmentWorkCommitmentService commitments,AppointmentWorkChanges changes) { this.commitments=commitments; this.changes=changes; }
private final FbaReceivingStore receiving;
public AppointmentWorkAppointmentChanges(AppointmentWorkCommitmentService commitments,AppointmentWorkChanges changes,FbaReceivingStore receiving) { this.commitments=commitments; this.changes=changes;this.receiving=receiving; }
public void changed(Object response) {
changes.requirementsChanged();
if(response instanceof FbaAppointmentResponse appointment) record(appointment);
@@ -20,9 +23,43 @@ public class AppointmentWorkAppointmentChanges {
else throw new IllegalArgumentException("预约命令缺少原业务响应");
}
private void record(FbaAppointmentResponse a) {
boolean cancelled=Set.of("CANCELLED","FAILED").contains(a.appointmentStatus());
var cargo=a.items().stream().map(i->List.of(i.customerId(),i.sourceKey(),Objects.toString(i.destinationKey(),""),Objects.toString(i.packageCount(),""),Objects.toString(i.shipmentReferences(),""))).sorted(Comparator.comparing(Object::toString)).toList();
commitments.record(a.batchKey(),new AppointmentWorkCommitmentService.Commitment(cancelled?null:target(a),cancelled?null:confirmed(a),null,AppointmentWorkPolicy.digest(cargo.toString()),cancelled));
var snapshot=receiving.snapshots(List.of(a.batchKey())).get(0);
for(var customer:a.items().stream().map(i->i.customerId()).distinct().sorted().toList()) {
var view=receiving(a,customer,snapshot);
var cargo=a.items().stream().filter(i->customer.equals(i.customerId())).map(i->List.of(i.customerId(),i.sourceKey(),Objects.toString(i.destinationKey(),""),Objects.toString(i.packageCount(),""),Objects.toString(i.shipmentReferences(),""))).sorted(Comparator.comparing(Object::toString)).toList();
commitments.record(AppointmentWorkPolicy.customerRequirement(a.batchKey(),customer),new AppointmentWorkCommitmentService.Commitment(view.target(),view.confirmed(),null,AppointmentWorkPolicy.digest(cargo.toString()),view.cancelled()));
}
}
record Receiving(Agreement agreement,Pending pending,Slot target,Slot confirmed,String timezone,boolean cancelled,String status) { }
static Receiving receiving(FbaAppointmentResponse batch,String customer,BatchReceiving snapshot) {
if(snapshot==null || !batch.batchKey().equals(snapshot.batchKey())) throw AppointmentWorkPolicy.unavailable("收货预约批量读取不完整");
var groups=snapshot.agreements().stream().filter(a->"ACTIVE".equals(a.status()) && a.members().stream().anyMatch(m->batch.batchKey().equals(m.batchKey()) && customer.equals(m.customerId()))).toList();
if(groups.size()>1) throw AppointmentWorkPolicy.unavailable("同一客户批次存在多个有效收货预约");
var waiting=snapshot.pendingChanges().stream().filter(p->"WAITING".equals(p.status()) && batch.batchKey().equals(p.intent().batchKey()) && customer.equals(p.intent().customerId())).toList();
if(waiting.size()>1) throw AppointmentWorkPolicy.unavailable("同一客户批次存在多个待确认变化");
var agreement=groups.isEmpty()?null:groups.get(0);
var plans=snapshot.plans().stream().filter(p->batch.batchKey().equals(p.batchKey()) && customer.equals(p.customerId())).toList();
if(plans.size()>1) throw AppointmentWorkPolicy.unavailable("同一客户批次存在多个独立计划");
boolean independentlyCovered=snapshot.agreements().stream().anyMatch(a->"ACTIVE".equals(a.status()) && a.members().stream().anyMatch(m->batch.batchKey().equals(m.batchKey())));
String status=null,timezone=null;Slot target=null,confirmed=null;
if(agreement!=null) {
var details=agreement.details();timezone=agreement.timezone();
if(details==null || !"SUCCESS".equals(details.appointmentStatus()) || details.appointmentAt()==null) throw AppointmentWorkPolicy.unavailable("有效收货预约缺少正式确认时间");
target=target(details,timezone);status=details.appointmentStatus();
confirmed=slot(details.appointmentAt(),details.appointmentEndAt(),timezone);
} else if(!plans.isEmpty()) {
var plan=plans.get(0);var details=plan.details();timezone=plan.timezone();
if(details==null || !Set.of("PENDING","FAILED").contains(Objects.toString(details.appointmentStatus(),"")) || details.confirmationNo()!=null || details.appointmentAt()!=null || details.appointmentEndAt()!=null)
throw AppointmentWorkPolicy.unavailable("客户独立计划包含无效状态或正式确认字段");
target=target(details,timezone);status=details.appointmentStatus();
} else if(!independentlyCovered) {
// Only a batch still without independent receiving coverage can supply its original shared plan.
status=batch.appointmentStatus();timezone=batch.timezone();target="CANCELLED".equals(status)?null:target(batch);
}
return new Receiving(agreement,waiting.isEmpty()?null:waiting.get(0),target,confirmed,timezone,"CANCELLED".equals(status),status);
}
private static Slot target(com.baiyue.lastleg.inboundplan.dto.FbaAppointmentDetails details,String timezone) {
return details.plannedAt()!=null?slot(details.plannedAt(),details.plannedEndAt(),timezone):details.plannedDate()==null?null:new Slot(details.plannedDate(),null,null,null,timezone);
}
public static Slot target(FbaAppointmentResponse a) {
return a.plannedAt()!=null?slot(a.plannedAt(),a.plannedEndAt(),a.timezone()):a.plannedDate()==null?null:new Slot(a.plannedDate(),null,null,null,a.timezone());
@@ -4,6 +4,8 @@ import com.baiyue.lastleg.appointmentplanning.*;
import com.baiyue.lastleg.appointmentwork.*;
import com.baiyue.lastleg.inboundplan.service.InboundFbaAppointmentService;
import com.baiyue.lastleg.inboundplan.dto.FbaAppointmentResponse;
import com.baiyue.lastleg.inboundplan.receiving.FbaReceivingStore;
import com.baiyue.lastleg.inboundplan.receiving.FbaReceivingContracts.BatchReceiving;
import com.baiyue.lastleg.security.OverseasPermissions;
import org.springframework.stereotype.Component;
import java.util.*;
@@ -19,8 +21,9 @@ public class AppointmentWorkDeliveryProvider implements AppointmentWorkFactGatew
private final AppointmentWorkCommitmentService commitments;
private final AppointmentWorkDeliveryMapper deliveries;
private final AppointmentWorkSourceMapper sourceMapping;
public AppointmentWorkDeliveryProvider(InboundFbaAppointmentService appointments,AppointmentCargoVerificationService cargo,AppointmentWorkCommitmentService commitments,AppointmentWorkDeliveryMapper deliveries,AppointmentWorkSourceMapper sourceMapping) {
this.appointments=appointments; this.cargo=cargo; this.commitments=commitments; this.deliveries=deliveries; this.sourceMapping=sourceMapping;
private final FbaReceivingStore receiving;
public AppointmentWorkDeliveryProvider(InboundFbaAppointmentService appointments,AppointmentCargoVerificationService cargo,AppointmentWorkCommitmentService commitments,AppointmentWorkDeliveryMapper deliveries,AppointmentWorkSourceMapper sourceMapping,FbaReceivingStore receiving) {
this.appointments=appointments; this.cargo=cargo; this.commitments=commitments; this.deliveries=deliveries; this.sourceMapping=sourceMapping;this.receiving=receiving;
}
public boolean supports(Type type,Flow flow) { return type==Type.DELIVERY && flow==Flow.TRANSFER; }
public List<Fact> read(List<FactRequest> requests) {
@@ -42,44 +45,55 @@ public class AppointmentWorkDeliveryProvider implements AppointmentWorkFactGatew
var snapshots=new HashMap<String,FbaAppointmentResponse>();
var deliveryRows=new HashMap<String,List<AppointmentWorkDeliveryMapper.Delivery>>();
Map<String,String> impacts=Map.of();
var receivingSnapshots=new HashMap<String,BatchReceiving>();
if(!batches.isEmpty()) {
appointments.batchSnapshots(batches).forEach(a->snapshots.put(a.batchKey(),a));
deliveries.batches(batches).forEach(d->deliveryRows.computeIfAbsent(d.batchKey(),ignored->new ArrayList<>()).add(d.delivery()));
impacts=commitments.impactTokens(batches);
receiving.snapshots(batches).forEach(s->receivingSnapshots.put(s.batchKey(),s));
impacts=commitments.impactTokens(requests.stream().filter(r->r.batchKey()!=null && r.batchKey().startsWith("FBA_APPOINTMENT:")).map(r->customerRequirement(r.batchKey(),deliveryCustomer(r.sources()))).distinct().toList());
}
var context=new BatchFacts(snapshots,deliveryRows,impacts);
var context=new BatchFacts(snapshots,deliveryRows,impacts,receivingSnapshots);
return requests.stream().map(r->read(r,privateSources.get(r.workItemKey()),inspections,context)).toList();
}
private record BatchFacts(Map<String,FbaAppointmentResponse> appointments,Map<String,List<AppointmentWorkDeliveryMapper.Delivery>> deliveries,Map<String,String> impacts) { }
private record BatchFacts(Map<String,FbaAppointmentResponse> appointments,Map<String,List<AppointmentWorkDeliveryMapper.Delivery>> deliveries,Map<String,String> impacts,Map<String,BatchReceiving> receiving) { }
private Fact read(FactRequest request,List<AppointmentWorkSourceMapper.Source> privateSources,Map<AppointmentCoverageService.SourceRef,AppointmentCargoVerificationService.Inspection> inspections,BatchFacts context) {
if(request.batchKey()==null) return privateSources!=null?customer(request,privateSources):unbatched(request,inspections);
if(request.batchKey().startsWith("DISPATCH:")) return customerBatch(request);
var a=context.appointments().get(request.batchKey());
var impact=context.impacts().get(request.batchKey());
var customer=deliveryCustomer(request.sources());
var impact=context.impacts().get(customerRequirement(request.batchKey(),customer));
if(a==null || impact==null) throw unavailable("预约批次事实批量读取不完整");
boolean cancelled=Set.of("CANCELLED","FAILED").contains(a.appointmentStatus());
var target=cancelled?null:target(a); var confirmed=cancelled?null:confirmed(a);
var rows=context.deliveries().getOrDefault(a.batchKey(),List.of());
if(rows.size()>1000) throw unavailable("批次派送事实超过完整读取上限");
var receiving=receiving(a,customer,context.receiving().get(a.batchKey()));
boolean unsuccessful=receiving.agreement()==null && (receiving.cancelled() || "FAILED".equals(receiving.status()));
var target=receiving.target();var confirmed=receiving.confirmed();
var customerItems=a.items().stream().filter(i->customer.equals(i.customerId())).toList();
var sourceRefs=customerItems.stream().map(i->new SourceReference(i.customerId(),i.sourceKey(),i.destinationKey())).distinct().toList();
if(customerItems.isEmpty() || !new HashSet<>(sourceRefs).equals(new HashSet<>(request.sources()))) throw conflict("当前客户批次来源已变化,请先同步工作清单");
var batchRows=context.deliveries().getOrDefault(a.batchKey(),List.of());
if(batchRows.size()>1000) throw unavailable("批次派送事实超过完整读取上限");
var rows=batchRows.stream().filter(d->customer.equals(d.customerId())).toList();
boolean complete=!rows.isEmpty() && rows.stream().allMatch(d->d.status()!=null && Set.of("DELIVERED","RETURN_PENDING","RETURNED").contains(d.status()) && d.signedAt()!=null && d.packages()!=null && d.packages().equals(d.signedPackages()));
// Every original source allocation must have real dispatch coverage, not just one signed fragment.
for(var item:a.items()) {
for(var item:customerItems) {
long allocated=rows.stream().filter(d->item.customerId().equals(d.customerId()) && item.sourceKey().equals(d.sourceKey()) && item.destinationKey().equals(d.destinationKey())).mapToLong(d->d.allocatedPackages()).sum();
if(item.packageCount()==null || allocated!=item.packageCount()) complete=false;
}
if(rows.stream().anyMatch(d->sourceRefs.stream().noneMatch(s->Objects.equals(s.sourceKey(),d.sourceKey()) && Objects.equals(s.destinationKey(),d.destinationKey())))) complete=false;
var actual=complete?new Actual(null,rows.stream().map(AppointmentWorkDeliveryMapper.Delivery::signedAt).max(Comparator.naturalOrder()).orElseThrow(),null):null;
var execution=complete?ExecutionStatus.COMPLETED:rows.stream().anyMatch(d->Set.of("DEPARTED","PARTIAL_DEPARTED").contains(d.dispatchStatus()))?ExecutionStatus.IN_PROGRESS:ExecutionStatus.PENDING;
var execution=complete?ExecutionStatus.COMPLETED:rows.stream().anyMatch(d->"DEPARTED".equals(d.dispatchStatus()) || "PARTIAL_DEPARTED".equals(d.dispatchStatus()))?ExecutionStatus.IN_PROGRESS:ExecutionStatus.PENDING;
var blocks=new ArrayList<String>();
if(cancelled) blocks.add("原预约已取消或失败,需求须重新安排");
if(unsuccessful) blocks.add("原预约已取消或未约成,需求须继续安排");
if(!a.replacementBatchKeys().isEmpty()) blocks.add("原批次已替代,请进入新批次");
var sourceRefs=a.items().stream().map(i->new SourceReference(i.customerId(),i.sourceKey(),i.destinationKey())).toList();
var action=complete?null:new Action(confirmed==null?ActionCode.RECORD_CONFIRMATION:ActionCode.CHANGE_BOOKING,confirmed==null?"维护原预约并登记确认":"核对或变更原预约",request.workItemKey(),new AppointmentTarget(a.id().toString(),a.batchKey(),sourceRefs));
if(receiving.pending()!=null) blocks.add("本客户收货范围变化待仓库明确回复,暂不可发车");
if(execution==ExecutionStatus.IN_PROGRESS) blocks.add("本客户运输已执行,普通改约已停止,请从原运输异常处理");
var action=execution!=ExecutionStatus.PENDING?null:new Action(receiving.pending()!=null?ActionCode.CHANGE_BOOKING:confirmed==null?ActionCode.RECORD_CONFIRMATION:ActionCode.CHANGE_BOOKING,receiving.pending()!=null?"继续处理待确认变化":unsuccessful?"沿原批次重新预约":confirmed==null?"维护原预约并登记确认":"核对或变更原预约",request.workItemKey(),new AppointmentTarget(a.id().toString(),a.batchKey(),sourceRefs));
var references=new ArrayList<BusinessReference>(); references.add(new BusinessReference("FBA_APPOINTMENT",a.batchKey(),a.appointmentNo(),"送仓预约"));
if(receiving.pending()!=null) references.add(new BusinessReference("RECEIVING_CHANGE",receiving.pending().changeKey(),null,"待仓库确认的变化"));
rows.stream().filter(d->d.deliveryNo()!=null).map(d->new BusinessReference("DELIVERY",d.deliveryNo(),d.deliveryNo(),"实际派送")).distinct().forEach(references::add);
return new Fact(request.workItemKey(),digest(a.batchKey()+":"+a.version()+":"+rows),impact,true,
confirmed!=null?PlanStatus.CONFIRMED:target!=null?PlanStatus.TARGET_SET:PlanStatus.UNPLANNED,execution,target,confirmed,actual,a.timezone(),a.appointmentNo(),null,
List.of(ORDER_READ),action==null?List.of():List.of(OverseasPermissions.INBOUND_ORDER_FBA_APPOINTMENT),action,blocks,
actual!=null?List.of("SOURCE_TIMEZONE"):List.of(),List.copyOf(references),null);
return new Fact(request.workItemKey(),digest(customerItems+":"+target+":"+confirmed+":"+receiving.pending()+":"+receiving.status()+":"+rows),impact,true,
receiving.pending()!=null?PlanStatus.AWAITING_REPLY:confirmed!=null?PlanStatus.CONFIRMED:target!=null?PlanStatus.TARGET_SET:PlanStatus.UNPLANNED,execution,target,confirmed,actual,receiving.timezone(),a.appointmentNo(),null,
List.of(ORDER_READ),action==null?List.of():receiving.pending()!=null?List.of(OverseasPermissions.INBOUND_ORDER_FBA_APPOINTMENT,"platform:appointment:delete"):List.of(OverseasPermissions.INBOUND_ORDER_FBA_APPOINTMENT),action,blocks,
actual!=null?List.of("SOURCE_TIMEZONE"):List.of(),List.copyOf(references),null,null);
}
private Fact unbatched(FactRequest request,Map<AppointmentCoverageService.SourceRef,AppointmentCargoVerificationService.Inspection> inspections) {
var blocks=new ArrayList<String>(); var missing=new ArrayList<String>(); Action action=null;
@@ -102,7 +116,7 @@ public class AppointmentWorkDeliveryProvider implements AppointmentWorkFactGatew
}
return new Fact(request.workItemKey(),digest(request.requirementKey()+":"+fingerprints),"UNBATCHED",false,PlanStatus.UNPLANNED,ExecutionStatus.PENDING,
null,null,null,null,null,null,List.of(ORDER_READ),action==null?List.of():List.of(OverseasPermissions.INBOUND_ORDER_FBA_APPOINTMENT),action,
blocks.stream().distinct().toList(),missing,List.of(),null);
blocks.stream().distinct().toList(),missing,List.of(),null,null);
}
private Fact customer(FactRequest request,List<AppointmentWorkSourceMapper.Source> sources) {
var rows=deliveries.customer(request.containerBookingId(),request.requirementKey().substring(8));
@@ -117,7 +131,7 @@ public class AppointmentWorkDeliveryProvider implements AppointmentWorkFactGatew
// removal explicitly so compensation can advance to that page and restore the original demand.
if(sources.isEmpty() || sources.stream().anyMatch(s->"CANCELLED".equals(s.status())))
return new Fact(request.workItemKey(),digest("REMOVED:"+sources),"REMOVED",false,PlanStatus.UNPLANNED,ExecutionStatus.PENDING,null,null,null,null,
sources.isEmpty()?null:sources.get(0).dispatchNo(),null,List.of(ORDER_READ),List.of(),null,List.of("原客户仓排车已取消或移除实物,待完整需求同步"),List.of(),List.of(),null);
sources.isEmpty()?null:sources.get(0).dispatchNo(),null,List.of(ORDER_READ),List.of(),null,List.of("原客户仓排车已取消或移除实物,待完整需求同步"),List.of(),List.of(),null,null);
return customerFact(request,deliveries.privateBatch(request.batchKey()),sources.stream().map(s->new Expected(s.orderId(),s.customerId(),s.packages())).toList(),sources.get(0).dispatchNo());
}
private Fact customerFact(FactRequest request,List<AppointmentWorkDeliveryMapper.CustomerDelivery> rows,List<Expected> sources,String dispatchNo) {
@@ -139,6 +153,6 @@ public class AppointmentWorkDeliveryProvider implements AppointmentWorkFactGatew
return new Fact(request.workItemKey(),digest(sources+":"+rows),digest(sources+":"+dates),!rows.isEmpty(),target==null?PlanStatus.UNPLANNED:PlanStatus.TARGET_SET,
complete?ExecutionStatus.COMPLETED:rows.isEmpty()?ExecutionStatus.PENDING:ExecutionStatus.IN_PROGRESS,target,null,actual,null,dispatchNo!=null?dispatchNo:rows.size()==1?rows.get(0).deliveryNo():null,null,
List.of(ORDER_READ),action==null?List.of():List.of("platform:overseas-warehouse:delivery:list"),action,
dates.size()>1?List.of("同一收货需求存在多个派送日期,请在原派送单核对"):List.of(),target!=null || actual!=null?List.of("SOURCE_TIMEZONE"):List.of(),List.copyOf(refs),null);
dates.size()>1?List.of("同一收货需求存在多个派送日期,请在原派送单核对"):List.of(),target!=null || actual!=null?List.of("SOURCE_TIMEZONE"):List.of(),List.copyOf(refs),null,null);
}
}
@@ -12,12 +12,33 @@ import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
@Component
@Transactional(propagation=Propagation.MANDATORY)
public class AppointmentWorkDispatchBinding implements AppointmentWorkDispatchPort {
@Transactional(readOnly=true,propagation=Propagation.SUPPORTS)
public Map<Long,List<BatchReference>> references(List<Long> dispatchIds) {
if(dispatchIds==null || dispatchIds.isEmpty()) return Map.of();
if(dispatchIds.stream().anyMatch(id->id==null || id<=0)) throw bad("排车引用缺少真实排车身份");
var rows=mapper.references(dispatchIds.stream().distinct().sorted().toList());
if(rows.size()>20000) throw unavailable("排车批次引用超过完整读取上限");
record Group(long dispatchId,String batchKey,String batchNo,String customerId) { }
var groups=new LinkedHashMap<Group,Map<List<String>,Integer>>();
for(var row:rows) {
if(!dispatchIds.contains(row.dispatchId()) || row.batchKey()==null || row.customerId()==null || row.sourceKey()==null || row.destinationKey()==null || row.packages()<=0)
throw unavailable("原排车逐票批次引用不完整");
groups.computeIfAbsent(new Group(row.dispatchId(),row.batchKey(),row.batchNo(),row.customerId()),ignored->new LinkedHashMap<>())
.merge(List.of(row.sourceKey(),row.destinationKey()),row.packages(),Math::addExact);
}
var result=new LinkedHashMap<Long,List<BatchReference>>();dispatchIds.forEach(id->result.put(id,new ArrayList<>()));
groups.forEach((group,sources)->result.get(group.dispatchId()).add(new BatchReference(group.batchKey(),group.batchNo(),group.customerId(),
sources.entrySet().stream().map(e->new BatchSource(e.getKey().get(0),e.getKey().get(1),e.getValue())).toList())));
result.replaceAll((id,refs)->List.copyOf(refs));return Collections.unmodifiableMap(result);
}
private final AppointmentWorkDispatchMapper mapper;
private final InboundFbaAppointmentService appointments;
private final AppointmentSourceService sources;
private final AppointmentWorkChanges changes;
public AppointmentWorkDispatchBinding(AppointmentWorkDispatchMapper mapper,InboundFbaAppointmentService appointments,AppointmentSourceService sources,AppointmentWorkChanges changes) {
this.mapper=mapper; this.appointments=appointments; this.sources=sources; this.changes=changes;
private final com.baiyue.lastleg.inboundplan.receiving.FbaReceivingStore receiving;
public AppointmentWorkDispatchBinding(AppointmentWorkDispatchMapper mapper,InboundFbaAppointmentService appointments,AppointmentSourceService sources,AppointmentWorkChanges changes,
com.baiyue.lastleg.inboundplan.receiving.FbaReceivingStore receiving) {
this.mapper=mapper; this.appointments=appointments; this.sources=sources; this.changes=changes;this.receiving=receiving;
}
public void bind(String customer,long dispatch,List<PhysicalItem> items,List<Allocation> allocations) {
if(allocations==null) throw bad("批次实物分配清单必填");
@@ -51,14 +72,9 @@ public class AppointmentWorkDispatchBinding implements AppointmentWorkDispatchPo
}
for(var batch:batches) {
var context=appointments.batchContext(batch); var appointment=context.appointment();
if(Set.of("CANCELLED","FAILED").contains(appointment.appointmentStatus()) || !appointment.replacementBatchKeys().isEmpty()) throw conflict("不能绑定取消或被替代的预约批次");
var grouped=allocations.stream().filter(a->batch.equals(a.batchKey())).collect(Collectors.groupingBy(a->a.sourceKey()+"/"+a.destinationKey()));
for(var group:grouped.values()) {
var ref=group.get(0);
var original=appointment.items().stream().filter(i->customer.equals(i.customerId()) && ref.sourceKey().equals(i.sourceKey()) && ref.destinationKey().equals(i.destinationKey())).findFirst().orElseThrow(()->conflict("实物来源不属于所选预约批次"));
long count=group.stream().mapToLong(Allocation::packageCount).sum();
if(original.packageCount()==null || mapper.allocated(batch,customer,ref.sourceKey(),ref.destinationKey(),dispatch)+count>original.packageCount()) throw conflict("所选实物超过该批次逐票分配量");
}
// Pre-arrangement reserves real cargo only. requireReceiving separately gates actual departure per customer.
if(!appointment.replacementBatchKeys().isEmpty()) throw conflict("不能绑定已被替代的预约批次");
requireCurrentAllocation(appointment,customer,dispatch,allocations.stream().filter(a->batch.equals(a.batchKey())).toList());
}
mapper.remove(dispatch); persisted.forEach(mapper::insert); changes.requirementsChanged();
}
@@ -82,7 +98,15 @@ public class AppointmentWorkDispatchBinding implements AppointmentWorkDispatchPo
if(remaining.values().stream().anyMatch(n->n!=0)) throw conflict("批次逐票分配必须与所选托盘实际货物完全一致");
if(item.packages()!=null && selected.stream().mapToLong(Allocation::packageCount).sum()!=item.packages()) throw conflict("逐票分配与所选托盘箱数不一致");
}
public void requireDeparture(String dispatchNo) {
public void requireDeparture(String dispatchNo) { checkDeparture(dispatchNo,true); }
public String departureBlockReason(String dispatchNo) {
try { checkDeparture(dispatchNo,false);return null; }
catch(org.springframework.web.server.ResponseStatusException error) {
if(error.getStatusCode().value()!=409) throw error;
return error.getReason();
}
}
private void checkDeparture(String dispatchNo,boolean locked) {
var rows=mapper.bindings(dispatchNo);
var linked=rows.stream().map(AppointmentWorkDispatchMapper.Bound::palletId).collect(Collectors.toSet());
for(var item:mapper.dispatchPhysical(dispatchNo)) if(item.planPalletId()!=null && !linked.contains(item.planPalletId())) {
@@ -91,8 +115,17 @@ public class AppointmentWorkDispatchBinding implements AppointmentWorkDispatchPo
}
for(var batch:rows.stream().map(AppointmentWorkDispatchMapper.Bound::batchKey).distinct().sorted().toList()) {
if(batch.startsWith("DISPATCH:")) continue;
lock(batch); var a=appointments.batchContext(batch).appointment();
if(!"SUCCESS".equals(a.appointmentStatus()) || a.appointmentAt()==null || !a.replacementBatchKeys().isEmpty()) throw conflict("关联预约尚未有效确认或已被替代,请重新核对批次后发车");
if(locked) lock(batch);
var a=appointments.batchContext(batch).appointment();
if(!a.replacementBatchKeys().isEmpty()) throw conflict("关联批次已被替代,请重新核对实物批次后发车");
var allocated=rows.stream().filter(row->batch.equals(row.batchKey())).toList();
var dispatchIds=allocated.stream().map(AppointmentWorkDispatchMapper.Bound::dispatchId).distinct().toList();
if(dispatchIds.size()!=1) throw conflict("排车批次关联不一致,请重新核对实物分配");
for(var customer:allocated.stream().map(AppointmentWorkDispatchMapper.Bound::customerId).distinct().toList()) {
requireReceiving(batch,customer,allocated.stream().filter(row->customer.equals(row.customerId())).toList());
requireCurrentAllocation(a,customer,dispatchIds.get(0),allocated.stream().filter(row->customer.equals(row.customerId()))
.map(row->new Allocation(row.palletId(),row.batchKey(),row.customerId(),row.sourceKey(),row.destinationKey(),row.packages())).toList());
}
}
for(var group:rows.stream().collect(Collectors.groupingBy(AppointmentWorkDispatchMapper.Bound::palletId)).values()) {
var first=group.get(0);
@@ -100,11 +133,39 @@ public class AppointmentWorkDispatchBinding implements AppointmentWorkDispatchPo
group.stream().map(a->new Allocation(a.palletId(),a.batchKey(),a.customerId(),a.sourceKey(),a.destinationKey(),a.packages())).toList(),true,true);
}
}
private void requireReceiving(String batch,String customer,List<AppointmentWorkDispatchMapper.Bound> rows) {
if(receiving.hasPending(batch,customer)) throw conflict("本客户有待仓库确认的变化,保留原安排但禁止发车,请先完成明确回复");
var agreement=receiving.currentAgreement(batch,customer);
if(agreement==null || !"ACTIVE".equals(agreement.status()) || agreement.details()==null || !"SUCCESS".equals(agreement.details().appointmentStatus())
|| agreement.details().appointmentAt()==null) throw conflict("本客户尚无有效收货确认,请先完成当前预约后发车");
var members=agreement.members().stream().filter(m->batch.equals(m.batchKey()) && customer.equals(m.customerId())).toList();
if(members.size()!=1) throw conflict("当前收货确认未唯一覆盖本客户批次,请重新核对");
for(var group:rows.stream().collect(Collectors.groupingBy(r->List.of(r.sourceKey(),r.destinationKey()))).values()) {
var first=group.get(0);
var sources=members.get(0).sources().stream().filter(s->first.sourceKey().equals(s.sourceKey()) && first.destinationKey().equals(s.destinationKey())).toList();
if(sources.size()!=1 || mapper.allocated(batch,customer,first.sourceKey(),first.destinationKey(),first.dispatchId())
+group.stream().mapToLong(AppointmentWorkDispatchMapper.Bound::packages).sum()>sources.get(0).packageCount())
throw conflict("当前收货确认未完整覆盖本次逐票实物,请重新核对");
}
}
private void requireCurrentAllocation(com.baiyue.lastleg.inboundplan.dto.FbaAppointmentResponse appointment,String customer,long dispatch,List<Allocation> allocations) {
var grouped=allocations.stream().collect(Collectors.groupingBy(a->List.of(a.sourceKey(),a.destinationKey())));
for(var group:grouped.values()) {
var ref=group.get(0);
var originals=appointment.items().stream().filter(i->customer.equals(i.customerId()) && ref.sourceKey().equals(i.sourceKey()) && ref.destinationKey().equals(i.destinationKey())).toList();
if(originals.size()!=1) throw conflict("实物来源不属于当前预约批次的唯一分配,请重新核对");
var original=originals.get(0);
long count=group.stream().mapToLong(Allocation::packageCount).sum();
if(group.stream().anyMatch(a->a.packageCount()<=0) || original.packageCount()==null
|| mapper.allocated(appointment.batchKey(),customer,ref.sourceKey(),ref.destinationKey(),dispatch)+count>original.packageCount())
throw conflict("所选实物超过当前批次逐票分配量,请重新核对");
}
}
private void requireNoBatch(String customer,long pallet) {
var physical=mapper.physical(pallet,customer).stream().filter(p->p.destinationKey()!=null && p.destinationKey().startsWith("FBA:")).toList();
var canonical=AppointmentWorkPhysicalSources.canonical(physical,sources);
var refs=physical.stream().map(p->new AppointmentCoverageService.SourceRef(customer,AppointmentWorkPhysicalSources.ref(canonical,p).sourceKey(),p.destinationKey())).distinct().toList();
if(!refs.isEmpty() && appointments.relatedBatchSnapshots(refs).stream().anyMatch(a->!Set.of("CANCELLED","FAILED").contains(a.appointmentStatus()) && a.replacementBatchKeys().isEmpty()))
if(!refs.isEmpty() && appointments.relatedBatchSnapshots(refs).stream().anyMatch(a->a.replacementBatchKeys().isEmpty()))
throw conflict("所选实物已有预约批次,必须明确逐批逐票分配后才能排车或发车");
}
public void removeItem(long dispatchId,long itemId) { mapper.removeItem(dispatchId,itemId); changes.requirementsChanged(); }
@@ -19,7 +19,8 @@ public class AppointmentWorkDispatchCandidates {
@Transactional(readOnly=true)
public Result read(String batch) {
var appointment=appointments.batchContext(batch).appointment();
if(Set.of("CANCELLED","FAILED").contains(appointment.appointmentStatus()) || !appointment.replacementBatchKeys().isEmpty()) return new Result(batch,appointment.version(),List.of(),List.of("原预约取消或替代,不能选择实物"));
// Receiving outcome does not release this batch's cargo or decide another customer's physical eligibility.
if(!appointment.replacementBatchKeys().isEmpty()) return new Result(batch,appointment.version(),List.of(),List.of("原批次已替代,不能选择实物"));
var candidates=mapper.candidates(appointment.id());
if(candidates.size()>500) throw conflict("本批实物超过500托,请先按原预约分批");
var output=new ArrayList<Candidate>();
@@ -10,6 +10,54 @@ public interface AppointmentWorkDispatchMapper {
record Bound(long dispatchId,long itemId,long palletId,String batchKey,String customerId,String sourceKey,String destinationKey,int packages) { }
record Candidate(long planPalletId,String palletNo,long planId,String planNo,String customerId,String sourceStage,boolean locked) { }
record DispatchPhysical(long itemId,Long planPalletId,String customerId,String sourceStage,Integer packages) { }
record TransportRow(long dispatchId,String dispatchNo,String customerId,String dispatchStatus,Integer pickedPackages,Integer shippedPackages,
java.time.LocalDateTime updatedAt,long itemId,long palletId,Integer itemPackages,String sourceStage,
String batchKey,String sourceKey,String destinationKey,int packages,String tripNo,Long tripVersion,String tripStatus,
Long outboundId,String outboundStatus,String taskStatus) { }
record ReceivingSource(String batchKey,String customerId,String sourceKey,String destinationKey,Integer packages) { }
record ReferenceRow(long dispatchId,String batchKey,String batchNo,String customerId,String sourceKey,String destinationKey,int packages) { }
@Select("""
<script>
SELECT a.dispatch_id,a.batch_key,CASE WHEN a.batch_key=CONCAT('DISPATCH:',d.id) THEN d.dispatch_no ELSE p.appointment_no END batch_no,
a.customer_id,a.source_key,a.destination_key,a.package_count packages
FROM ll_dispatch_appointment_allocation a JOIN ll_dispatch_order d ON d.id=a.dispatch_id AND d.customer_id=a.customer_id
LEFT JOIN ll_inbound_fba_appointment p ON a.batch_key=CONCAT('FBA_APPOINTMENT:',p.id)
WHERE a.dispatch_id IN <foreach collection='ids' item='id' open='(' separator=',' close=')'>#{id}</foreach>
ORDER BY a.dispatch_id,a.batch_key,a.customer_id,a.source_key,a.destination_key,a.dispatch_item_id LIMIT 20001
</script>
""")
List<ReferenceRow> references(@Param("ids") List<Long> ids);
@Select("""
<script>
SELECT CONCAT('FBA_APPOINTMENT:',i.appointment_id) batch_key,i.customer_id,i.source_key,
CONCAT('FBA:',i.fba_warehouse_code) destination_key,i.package_count packages
FROM ll_inbound_fba_appointment_item i
WHERE i.active=1 AND CONCAT('FBA_APPOINTMENT:',i.appointment_id) IN
<foreach collection='batches' item='batch' open='(' separator=',' close=')'>#{batch}</foreach>
ORDER BY i.appointment_id,i.customer_id,i.source_key LIMIT 20001
</script>
""")
@Options(flushCache=Options.FlushCachePolicy.TRUE,useCache=false)
List<ReceivingSource> receivingSources(@Param("batches") List<String> batches);
@Select("""
<script>
SELECT a.dispatch_id,d.dispatch_no,a.customer_id,d.status dispatch_status,d.picked_package_count picked_packages,
d.shipped_package_count shipped_packages,d.updated_at,a.dispatch_item_id item_id,a.plan_pallet_id pallet_id,
i.package_count item_packages,i.source_stage,a.batch_key,a.source_key,a.destination_key,a.package_count packages,
t.trip_no,t.record_version trip_version,t.status trip_status,o.id outbound_id,o.status outbound_status,p.status task_status
FROM ll_dispatch_appointment_allocation a JOIN ll_dispatch_order d ON d.id=a.dispatch_id AND d.customer_id=a.customer_id
JOIN ll_dispatch_item i ON i.id=a.dispatch_item_id AND i.dispatch_order_id=d.id AND i.customer_id=d.customer_id
LEFT JOIN ll_fleet_trip_dispatch b ON b.dispatch_id=d.id LEFT JOIN ll_fleet_trip t ON t.id=b.trip_id
LEFT JOIN ll_outbound_order o ON o.dispatch_order_id=d.id AND o.customer_id=d.customer_id
LEFT JOIN ll_dispatch_pda_task p ON p.dispatch_order_id=d.id AND p.customer_id=d.customer_id
WHERE d.status&lt;&gt;'CANCELLED' AND (
a.dispatch_id IN (SELECT selected.dispatch_id FROM ll_dispatch_appointment_allocation selected WHERE selected.batch_key IN <foreach collection='batches' item='key' open='(' separator=',' close=')'>#{key}</foreach>)
<if test='trips != null and !trips.isEmpty()'>OR t.trip_no IN <foreach collection='trips' item='trip' open='(' separator=',' close=')'>#{trip}</foreach></if>
) ORDER BY a.dispatch_id,a.dispatch_item_id,a.batch_key,a.customer_id,a.source_key,a.destination_key LIMIT 1001
</script>
""")
@Options(flushCache=Options.FlushCachePolicy.TRUE,useCache=false)
List<TransportRow> transportRows(@Param("batches") List<String> batches,@Param("trips") List<String> trips);
@Select("SELECT i.id item_id,i.plan_pallet_id,i.customer_id,i.source_stage,i.package_count packages FROM ll_dispatch_item i JOIN ll_dispatch_order d ON d.id=i.dispatch_order_id WHERE d.dispatch_no=#{number}") List<DispatchPhysical> dispatchPhysical(String number);
@Select("SELECT DISTINCT p.id plan_pallet_id,p.pallet_no,p.plan_id,n.plan_no,p.customer_id,CASE WHEN p.warehouse_pallet_id IS NULL THEN 'PLAN' ELSE 'RECEIVED' END source_stage,EXISTS(SELECT 1 FROM ll_dispatch_pallet_lock k WHERE k.plan_pallet_id=p.id AND k.status='ACTIVE') locked FROM ll_inbound_plan_pallet p JOIN ll_inbound_plan n ON n.id=p.plan_id JOIN ll_inbound_plan_pallet_allocation a ON a.pallet_id=p.id JOIN ll_inbound_plan_line l ON l.id=a.line_id JOIN ll_inbound_fba_appointment_item i ON i.order_id=l.order_id AND i.customer_id=l.customer_id WHERE i.appointment_id=#{id} ORDER BY p.id LIMIT 501")
List<Candidate> candidates(long id);
@@ -40,6 +40,6 @@ public class AppointmentWorkReceivingProvider implements AppointmentWorkFactGate
String impact=digest(dates.toString());
return new Fact(request.workItemKey(),fingerprint,impact,!plans.isEmpty(),target==null?PlanStatus.UNPLANNED:PlanStatus.TARGET_SET,
execution,target,null,actual,null,plans.size()==1?plans.get(0).getPlanNo():null,null,List.of(OverseasPermissions.INBOUND_ORDER_LIST),
action==null?List.of():List.of(permission),action,List.copyOf(blocks),target!=null || actual!=null?List.of("SOURCE_TIMEZONE"):List.of(),refs,null);
action==null?List.of():List.of(permission),action,List.copyOf(blocks),target!=null || actual!=null?List.of("SOURCE_TIMEZONE"):List.of(),refs,null,null);
}
}
@@ -69,7 +69,20 @@ public class AppointmentWorkRequirementProvider implements AppointmentWorkFactGa
batch.fbaWarehouseCode(),batch.timezone(),references.stream().distinct().toList(),List.copyOf(replacementKeys),batchSources));
}
}
return List.copyOf(result.values());
var scoped=new ArrayList<Requirement>();
for(var requirement:result.values()) {
if(requirement.batchKey()==null || !requirement.batchKey().startsWith("FBA_APPOINTMENT:")) { scoped.add(requirement);continue; }
var customers=new TreeMap<String,List<SourceReference>>();
requirement.sources().forEach(s->customers.computeIfAbsent(s.customerId(),ignored->new ArrayList<>()).add(s));
if(customers.isEmpty()) throw unavailable("实物批次缺少客户来源,不能生成送仓事项");
customers.forEach((customer,refs)->{
var replacements=new LinkedHashSet<>(requirement.replacesRequirementKeys());
requirement.replacesRequirementKeys().stream().filter(k->k.startsWith("FBA_APPOINTMENT:")).map(k->customerRequirement(k,customer)).forEach(replacements::add);
scoped.add(new Requirement(customerRequirement(requirement.batchKey(),customer),requirement.batchKey(),requirement.batchNo(),requirement.destinationKind(),
requirement.originName(),requirement.destinationName(),requirement.timezone(),requirement.references(),List.copyOf(replacements),List.copyOf(refs)));
});
}
return List.copyOf(scoped);
}
private void customer(Requirement requirement,List<AppointmentWorkSourceMapper.Source> orders,List<AppointmentWorkSourceMapper.DispatchSource> dispatches,Map<String,Requirement> result) {
if(orders.stream().anyMatch(s->!"PRIVATE".equals(s.destinationType()))) throw conflict("客户仓需求与原订单目的地不一致");
@@ -106,6 +119,8 @@ public class AppointmentWorkRequirementProvider implements AppointmentWorkFactGa
return result.stream().distinct().toList();
}
private static boolean active(FbaAppointmentResponse batch) {
return !Set.of("CANCELLED","FAILED").contains(batch.appointmentStatus()) && (batch.replacementBatchKeys()==null || batch.replacementBatchKeys().isEmpty());
// Cancelling a booking clears its schedule, not its physical allocation. Only an
// authoritative replacement retires the batch and its delivery/vehicle obligations.
return batch.replacementBatchKeys()==null || batch.replacementBatchKeys().isEmpty();
}
}
@@ -0,0 +1,116 @@
package com.baiyue.lastleg.inboundplan.work;
import com.baiyue.lastleg.fleet.trip.*;
import com.baiyue.lastleg.inboundplan.receiving.FbaAppointmentExecutionPort;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.*;
import org.springframework.web.server.ResponseStatusException;
import java.time.Instant;
import java.util.*;
import java.util.stream.Collectors;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
/** Original allocations and signed/returned delivery evidence; never reads work projections. */
@Component
public class FbaAppointmentExecutionAdapter implements FbaAppointmentExecutionPort {
private final AppointmentWorkDispatchMapper allocations;
private final FleetTripMapper trips;
private final FleetDispatchGateway dispatches;
private record Facts(List<String> batches,List<AppointmentWorkDispatchMapper.ReceivingSource> sources,
List<AppointmentWorkDispatchMapper.TransportRow> rows,
Map<String,FleetCompletionModels.DispatchCompletion> outcomes,
Map<String,FleetDispatchGateway.TerminalEvidence> evidence,Map<String,String> invalid) { }
public FbaAppointmentExecutionAdapter(AppointmentWorkDispatchMapper allocations,FleetTripMapper trips,FleetDispatchGateway dispatches) {
this.allocations=allocations;this.trips=trips;this.dispatches=dispatches;
}
@Transactional(readOnly=true)
public Scope inspect(List<Member> input) {
var members=members(input);return scope(members,read(batches(members)));
}
@Transactional(propagation=Propagation.MANDATORY)
public Scope lockAndInspect(List<Member> input,String expectedVersion) {
var members=members(input);if(members.isEmpty()) throw bad("请选择需要重新预约的客户");
var before=read(batches(members));var selected=selected(before,members);
var tripNos=tripNos(selected);var numbers=dispatchNos(selected);
for(var trip:tripNos) if(trips.lockByNo(trip)==null) throw conflict("原车次关系已变化,请重新核对");
for(var number:numbers) {
dispatches.completion(number,true);
dispatches.lockForPlanning(number);
}
var current=read(before.batches());var scope=scope(members,current);
if(!numbers.equals(dispatchNos(selected(current,members))) || !tripNos.equals(tripNos(selected(current,members)))
|| expectedVersion==null || !expectedVersion.equals(scope.version())) throw conflict("客户执行范围已变化,请重新读取预约候选");
for(var member:scope.members()) if(!member.canRebook()) throw conflict(member.reason());
return scope;
}
private Facts read(List<String> batches) {
if(batches.isEmpty()) return new Facts(batches,List.of(),List.of(),Map.of(),Map.of(),Map.of());
var sources=allocations.receivingSources(batches);var rows=allocations.transportRows(batches,List.of());
if(sources.size()>20000 || rows.size()>1000) throw unavailable("客户执行事实超过完整读取上限");
var outcomes=new TreeMap<String,FleetCompletionModels.DispatchCompletion>();var evidence=new TreeMap<String,FleetDispatchGateway.TerminalEvidence>();var invalid=new TreeMap<String,String>();
for(var number:dispatchNos(rows)) {
var completion=dispatches.completion(number,false);
if(completion==null || !number.equals(completion.dispatchNo()) || completion.outcomes()==null) throw unavailable("原派送执行事实读取不完整");
outcomes.put(number,completion);
if(rows.stream().anyMatch(r->number.equals(r.dispatchNo()) && "DEPARTED".equals(r.dispatchStatus()))) {
try {
var fact=dispatches.terminalEvidence(number,false);
if(fact==null || !number.equals(fact.dispatchNo()) || fact.deliveries()==null) throw unavailable("原发运签收退回事实缺失");
evidence.put(number,fact);
} catch(ResponseStatusException error) {
if(error.getStatusCode().value()!=409) throw error;
invalid.put(number,Objects.toString(error.getReason(),"原发运证据不完整"));
}
}
}
return new Facts(batches,List.copyOf(sources),List.copyOf(rows),outcomes,evidence,invalid);
}
private Scope scope(List<Member> members,Facts facts) {
// The selected customer subset is deliberately absent from the version. All facts for the
// same batch set participate, including siblings omitted from the candidate selection.
return new Scope(digest(facts.toString()),members.stream().map(m->eligibility(m,facts)).toList());
}
private Eligibility eligibility(Member member,Facts facts) {
var expected=facts.sources().stream().filter(s->member.batchKey().equals(s.batchKey()) && member.customerId().equals(s.customerId())).toList();
if(expected.isEmpty() || expected.stream().anyMatch(s->s.packages()==null || s.packages()<=0)) return denied(member,false,"客户已移出本批或来源箱数尚未核实");
var rows=selected(facts,List.of(member));
boolean complete=!rows.isEmpty();
var expectedAmounts=new HashMap<List<String>,Long>();
for(var source:expected) if(source.sourceKey()==null || source.destinationKey()==null || expectedAmounts.put(List.of(source.sourceKey(),source.destinationKey()),source.packages().longValue())!=null)
return denied(member,false,"原客户逐票分配存在重复或缺失");
var allocated=rows.stream().collect(Collectors.groupingBy(r->List.of(r.sourceKey(),r.destinationKey()),Collectors.summingLong(AppointmentWorkDispatchMapper.TransportRow::packages)));
if(rows.stream().anyMatch(r->r.packages()<=0) || allocated.entrySet().stream().anyMatch(e->!expectedAmounts.containsKey(e.getKey()) || e.getValue()>expectedAmounts.get(e.getKey())))
return denied(member,false,"排车逐票数量或来源不属于当前客户批次,请先核对原实物关联");
if(!expectedAmounts.equals(allocated)) complete=false;
for(var row:rows) {
var proof=facts.evidence().get(row.dispatchNo());
if(proof==null || !member.customerId().equals(proof.customerId())) { complete=false;continue; }
var delivery=proof.deliveries().stream().filter(d->member.customerId().equals(d.customerId()) && row.destinationKey().equals(d.destinationKey())).toList();
long dispatched=facts.rows().stream().filter(r->row.dispatchNo().equals(r.dispatchNo()) && row.customerId().equals(r.customerId()) && row.destinationKey().equals(r.destinationKey())).mapToLong(AppointmentWorkDispatchMapper.TransportRow::packages).sum();
if(delivery.size()!=1 || delivery.get(0).packages()!=dispatched || delivery.get(0).terminalAt()==null || delivery.get(0).terminalAt().isAfter(Instant.now())) complete=false;
}
if(complete) return denied(member,true,"本客户货物已实际签收或完成退回入仓,不再重新预约");
for(var row:rows) {
if(facts.invalid().containsKey(row.dispatchNo())) return denied(member,false,facts.invalid().get(row.dispatchNo()));
var outcomes=facts.outcomes().get(row.dispatchNo()).outcomes();
if(outcomes.stream().anyMatch(o->o.terminalAt()!=null || o.signedPackages()>0 || o.returnedPackages()>0 || Set.of("DELIVERING","DELIVERED","REJECTED","RETURN_PENDING","RETURNED").contains(Objects.toString(o.status(),""))))
return denied(member,false,"本客户已有实际派送执行,请从原运输异常处理");
if(!"PENDING_DEPARTURE".equals(row.dispatchStatus()) || positive(row.pickedPackages()) || positive(row.shippedPackages())
|| row.taskStatus()!=null && !"PDA_PENDING".equals(row.taskStatus()) || row.outboundStatus()!=null && !"WAIT_OUTBOUND".equals(row.outboundStatus())
|| row.tripNo()!=null && !"PLANNED".equals(row.tripStatus())) return denied(member,false,"本客户运输已开始执行,不能普通重新预约");
}
return new Eligibility(member,false,true,null);
}
private static Eligibility denied(Member member,boolean completed,String reason) { return new Eligibility(member,completed,false,reason); }
private static boolean positive(Integer value) { return value!=null && value>0; }
private static List<AppointmentWorkDispatchMapper.TransportRow> selected(Facts facts,List<Member> members) {
var scopes=new HashSet<>(members);return facts.rows().stream().filter(r->scopes.contains(new Member(r.batchKey(),r.customerId()))).toList();
}
private static List<String> dispatchNos(List<AppointmentWorkDispatchMapper.TransportRow> rows) { return rows.stream().map(AppointmentWorkDispatchMapper.TransportRow::dispatchNo).distinct().sorted().toList(); }
private static List<String> tripNos(List<AppointmentWorkDispatchMapper.TransportRow> rows) { return rows.stream().map(AppointmentWorkDispatchMapper.TransportRow::tripNo).filter(Objects::nonNull).distinct().sorted().toList(); }
private static List<String> batches(List<Member> members) { return members.stream().map(Member::batchKey).distinct().sorted().toList(); }
private static List<Member> members(List<Member> members) {
if(members==null || members.size()>200 || new HashSet<>(members).size()!=members.size() || members.stream().anyMatch(m->m==null || m.batchKey()==null || !m.batchKey().matches("FBA_APPOINTMENT:[1-9][0-9]*") || m.customerId()==null || m.customerId().isBlank())) throw bad("客户执行范围须为最多200个不重复真实客户批次");
return members.stream().sorted(Comparator.comparing(Member::batchKey).thenComparing(Member::customerId)).toList();
}
}
@@ -0,0 +1,159 @@
package com.baiyue.lastleg.inboundplan.work;
import com.baiyue.lastleg.fleet.trip.*;
import com.baiyue.lastleg.inboundplan.receiving.FbaAppointmentTransportPort;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.*;
import org.springframework.transaction.support.*;
import java.util.*;
import java.util.stream.Collectors;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
/** Mutates original dispatch rows and fleet bindings only; owns no cargo quantity ledger. */
@Component
public class FbaAppointmentTransportAdapter implements FbaAppointmentTransportPort {
private final AppointmentWorkDispatchMapper allocations;
private final FleetTripMapper trips;
private final FleetDispatchGateway dispatches;
private final FleetTripService fleet;
private final FleetBookingReservations reservations;
private final com.baiyue.lastleg.appointmentwork.AppointmentWorkChanges changes;
private final Object lockContextKey=new Object();
private record Locked(List<String> batches,List<String> tripNos,List<AppointmentWorkDispatchMapper.TransportRow> rows,Scope scope) { }
private record Source(String batch,String customer,String source,String destination) { }
public FbaAppointmentTransportAdapter(AppointmentWorkDispatchMapper allocations,FleetTripMapper trips,
FleetDispatchGateway dispatches,@Lazy FleetTripService fleet,FleetBookingReservations reservations,
com.baiyue.lastleg.appointmentwork.AppointmentWorkChanges changes) {
this.allocations=allocations;this.trips=trips;this.dispatches=dispatches;this.fleet=fleet;this.reservations=reservations;this.changes=changes;
}
@Transactional(readOnly=true)
public Scope inspect(List<String> batchKeys) {
var batches=keys(batchKeys);var first=read(batches,List.of());
return scope(batches,read(batches,tripNos(first)));
}
@Transactional(propagation=Propagation.MANDATORY)
public Scope lockAndInspect(List<String> batchKeys,String expectedVersion) {
var batches=keys(batchKeys);var initial=read(batches,List.of());var tripNos=tripNos(initial);
for(var trip:tripNos) if(trips.lockByNo(trip)==null) throw conflict("原车次已变化,请重新核对");
var all=read(batches,tripNos);
var targets=all.stream().filter(r->batches.contains(r.batchKey())).map(AppointmentWorkDispatchMapper.TransportRow::dispatchNo).distinct().sorted().toList();
for(var number:targets) dispatches.lockForPlanning(number);
var lockedRows=read(batches,tripNos);
if(!tripNos(lockedRows).equals(tripNos)) throw conflict("排车车次关联已变化,请重新核对");
var scope=scope(batches,lockedRows);
if(expectedVersion==null || !expectedVersion.equals(scope.version())) throw conflict("运输范围或版本已变化,请重新读取调整范围");
if(!scope.blockReasons().isEmpty()) throw conflict(String.join(";",scope.blockReasons()));
if(TransactionSynchronizationManager.hasResource(lockContextKey)) throw conflict("同一事务不能重复打开运输调整范围");
TransactionSynchronizationManager.bindResource(lockContextKey,new Locked(batches,tripNos,lockedRows,scope));
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
public void afterCompletion(int status) { TransactionSynchronizationManager.unbindResourceIfPossible(lockContextKey); }
});
return scope;
}
@Transactional(propagation=Propagation.MANDATORY)
public Result apply(Scope locked,List<Move> moves,String operator,String reason) {
var context=(Locked)TransactionSynchronizationManager.getResource(lockContextKey);
if(context==null || !context.scope().equals(locked)) throw conflict("运输调整必须使用本事务锁定的权威范围");
if(operator==null || operator.isBlank() || reason==null || reason.isBlank()) throw bad("运输调整须提供操作人和原因");
var current=read(context.batches(),context.tripNos());
if(!scope(context.batches(),current).version().equals(locked.version())) throw conflict("锁定后的排车事实已变化,请重新核对");
var intents=new LinkedHashMap<Source,Move>();
if(moves==null || moves.size()>1000) throw bad("运输调整来源清单无效");
for(var move:moves) {
if(move==null || !context.batches().contains(move.batchKey()) || move.originalPackages()<=0 || move.retainedPackages()<0 || move.retainedPackages()>move.originalPackages()
|| move.customerId()==null || move.sourceKey()==null || move.destinationKey()==null) throw bad("运输调整必须对应原批次及0至当前量的真实逐票分配");
if(move.retainedPackages()<move.originalPackages() && (move.laterBatchKey()==null || !move.laterBatchKey().startsWith("FBA_APPOINTMENT:"))) throw bad("移出货物缺少核心生成的后送批次");
var source=new Source(move.batchKey(),move.customerId(),move.sourceKey(),move.destinationKey());
if(intents.putIfAbsent(source,move)!=null) throw bad("运输调整来源重复");
}
var targetRows=current.stream().filter(r->intents.containsKey(source(r))).toList();
var counts=targetRows.stream().collect(Collectors.groupingBy(this::source,Collectors.summingLong(AppointmentWorkDispatchMapper.TransportRow::packages)));
var remove=new LinkedHashMap<Source,Integer>();
for(var entry:intents.entrySet()) {
long bound=counts.getOrDefault(entry.getKey(),0L);
if(bound>entry.getValue().originalPackages()) throw conflict("原排车实物超过核心批次当前量,不能调整");
remove.put(entry.getKey(),Math.toIntExact(Math.max(0,bound-entry.getValue().retainedPackages())));
}
var byItem=current.stream().collect(Collectors.groupingBy(AppointmentWorkDispatchMapper.TransportRow::itemId,TreeMap::new,Collectors.toList()));
var selected=selectWholeItems(byItem,remove);
var affected=targetRows.stream().filter(r->selected.contains(r.itemId())).map(AppointmentWorkDispatchMapper.TransportRow::dispatchNo).distinct().sorted().toList();
// Validate every partial edit before invoking any original command. Transaction rollback is still mandatory.
for(var number:affected) {
var rows=current.stream().filter(r->number.equals(r.dispatchNo())).toList();
boolean whole=rows.stream().allMatch(r->selected.contains(r.itemId()));
var physical=allocations.dispatchPhysical(number);
if(physical.stream().anyMatch(p->!byItem.containsKey(p.itemId()))) throw conflict("原排车含未纳入逐票批次的实物,请先核对完整关联");
for(var item:physical) if(item.packages()==null || byItem.get(item.itemId()).stream().mapToLong(AppointmentWorkDispatchMapper.TransportRow::packages).sum()!=item.packages())
throw conflict("逐票批次分配未完整核平原排车实物,不能调整");
if(!whole && rows.stream().anyMatch(r->r.outboundId()!=null)) throw conflict("部分移出前须先在原业务处理未执行出库单");
}
var next=new LinkedHashSet<String>();
for(var number:affected) {
var rows=current.stream().filter(r->number.equals(r.dispatchNo())).toList();var first=rows.get(0);
boolean whole=rows.stream().allMatch(r->selected.contains(r.itemId()));
if(first.tripNo()!=null) fleet.unbindDispatch(first.tripNo(),operator,new FleetTripModels.BindingCommand(trips.findByNo(first.tripNo()).getRecordVersion(),number,reason));
if(whole) {
allocations.remove(first.dispatchId());
for(var row:rows) allocations.insert(new AppointmentWorkDispatchMapper.Bound(row.dispatchId(),row.itemId(),row.palletId(),intents.get(source(row)).laterBatchKey(),row.customerId(),row.sourceKey(),row.destinationKey(),row.packages()));
next.add("沿原排车重新安排后送车辆");
} else {
for(var item:rows.stream().filter(r->selected.contains(r.itemId())).map(AppointmentWorkDispatchMapper.TransportRow::itemId).distinct().sorted().toList()) dispatches.removeItem(number,item);
if(first.tripNo()!=null) fleet.bindDispatch(first.tripNo(),operator,new FleetTripModels.BindingCommand(trips.findByNo(first.tripNo()).getRecordVersion(),number,reason));
next.add("按后送批次重新选择已移出的实物");
}
}
for(var number:context.tripNos()) {
var detail=fleet.detail(number);
if(detail.dispatches().isEmpty() && !reservations.hasConfirmed(trips.findByNo(number).getId())) {
// The original cancellation command also checks external reservations and releases vehicle/driver resources.
fleet.cancel(number,operator,new FleetTripModels.Cancel(detail.trip().recordVersion(),reason));
next.add("原车无有效货物,已按车辆业务取消");
} else next.add("核对原车剩余货物及发车安排");
}
if(!affected.isEmpty()) changes.requirementsChanged();
var follow=new LinkedHashSet<>(context.batches());intents.values().stream().map(Move::laterBatchKey).filter(Objects::nonNull).forEach(follow::add);
return new Result(vehicles(read(List.copyOf(follow),context.tripNos())),List.copyOf(next));
}
private Set<Long> selectWholeItems(Map<Long,List<AppointmentWorkDispatchMapper.TransportRow>> byItem,Map<Source,Integer> wanted) {
var candidates=byItem.entrySet().stream().filter(e->e.getValue().stream().allMatch(r->wanted.getOrDefault(source(r),0)>0)).toList();
var selected=new LinkedHashSet<Long>();var budget=new int[]{10000};
if(!select(candidates,0,new HashMap<>(wanted),selected,budget)) throw conflict("移出数量不能精确匹配完整实物明细,请先在原库存业务按实际货物拆托后重试");
return selected;
}
private boolean select(List<Map.Entry<Long,List<AppointmentWorkDispatchMapper.TransportRow>>> items,int index,Map<Source,Integer> remaining,Set<Long> chosen,int[] budget) {
if(remaining.values().stream().allMatch(n->n==0)) return true;
if(index>=items.size() || --budget[0]<0) return false;
var item=items.get(index);var amounts=item.getValue().stream().collect(Collectors.groupingBy(this::source,Collectors.summingInt(AppointmentWorkDispatchMapper.TransportRow::packages)));
if(amounts.entrySet().stream().allMatch(e->e.getValue()<=remaining.getOrDefault(e.getKey(),0))) {
amounts.forEach((k,v)->remaining.compute(k,(ignored,n)->n-v));chosen.add(item.getKey());
if(select(items,index+1,remaining,chosen,budget)) return true;
chosen.remove(item.getKey());amounts.forEach((k,v)->remaining.compute(k,(ignored,n)->n+v));
}
return select(items,index+1,remaining,chosen,budget);
}
private Source source(AppointmentWorkDispatchMapper.TransportRow row) { return new Source(row.batchKey(),row.customerId(),row.sourceKey(),row.destinationKey()); }
private List<String> keys(List<String> keys) {
if(keys==null || keys.isEmpty() || keys.size()>100 || keys.stream().anyMatch(k->k==null || !k.matches("FBA_APPOINTMENT:[1-9][0-9]*"))) throw bad("请选择1至100个真实预约批次");
return keys.stream().distinct().sorted().toList();
}
private List<AppointmentWorkDispatchMapper.TransportRow> read(List<String> batches,List<String> tripNos) {
var rows=allocations.transportRows(batches,tripNos);if(rows.size()>1000) throw conflict("运输实物范围超过完整核对上限,请缩小范围");return rows;
}
private List<String> tripNos(List<AppointmentWorkDispatchMapper.TransportRow> rows) { return rows.stream().map(AppointmentWorkDispatchMapper.TransportRow::tripNo).filter(Objects::nonNull).distinct().sorted().toList(); }
private Scope scope(List<String> batches,List<AppointmentWorkDispatchMapper.TransportRow> rows) {
var reasons=new LinkedHashSet<String>();
for(var row:rows) if(batches.contains(row.batchKey())) {
if(!"PENDING_DEPARTURE".equals(row.dispatchStatus()) || row.pickedPackages()!=null && row.pickedPackages()>0 || row.shippedPackages()!=null && row.shippedPackages()>0
|| row.taskStatus()!=null && !"PDA_PENDING".equals(row.taskStatus()) || row.outboundStatus()!=null && !"WAIT_OUTBOUND".equals(row.outboundStatus())) reasons.add("排车已开始执行,不能普通移出;请进入原运输异常处理");
if(row.tripNo()!=null && !"PLANNED".equals(row.tripStatus())) reasons.add("原车次已开始执行,不能普通移出");
if(row.itemPackages()==null || row.packages()<=0) reasons.add("原排车实物数量不完整");
}
return new Scope(digest(batches+":"+rows),vehicles(rows),List.copyOf(reasons));
}
private List<Vehicle> vehicles(List<AppointmentWorkDispatchMapper.TransportRow> rows) {
return rows.stream().collect(Collectors.groupingBy(r->List.of(r.dispatchNo(),r.batchKey()),LinkedHashMap::new,Collectors.toList())).values().stream().map(group->{
var r=group.get(0);return new Vehicle(r.tripNo(),r.tripVersion()==null?null:Math.toIntExact(r.tripVersion()),r.dispatchNo(),r.customerId(),r.batchKey(),r.dispatchStatus(),group.stream().mapToInt(AppointmentWorkDispatchMapper.TransportRow::packages).sum());
}).toList();
}
}
@@ -0,0 +1,25 @@
package com.baiyue.lastleg.inboundplan.work;
import com.baiyue.lastleg.appointmentwork.AppointmentWorkFollowupInheritance;
import com.baiyue.lastleg.inboundplan.receiving.FbaAppointmentWorkPort;
import org.springframework.stereotype.Component;
import java.time.LocalDate;
import java.util.List;
@Component
public class FbaAppointmentWorkAdapter implements FbaAppointmentWorkPort {
private final AppointmentWorkFollowupInheritance work;
public FbaAppointmentWorkAdapter(AppointmentWorkFollowupInheritance work) { this.work=work; }
public Followup resolveFollowup(String ownerId,LocalDate date) {
var resolved=work.resolve(ownerId,date);
return new Followup(resolved.owner().id(),resolved.owner().name(),resolved.nextFollowupDate());
}
public void seedFollowup(Member original,Member later,Followup followup,String operatorId) {
work.seed(member(original),member(later),new com.baiyue.lastleg.appointmentwork.AppointmentWorkStore.Followup(
new com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.Owner(followup.ownerId(),followup.ownerName()),followup.nextFollowupDate()),operatorId);
}
public void inheritFollowup(List<Member> originals,List<Member> successors,String operatorId) {
work.inherit(originals.stream().map(FbaAppointmentWorkAdapter::member).toList(),successors.stream().map(FbaAppointmentWorkAdapter::member).toList(),operatorId);
}
private static AppointmentWorkFollowupInheritance.Member member(Member member) { return new AppointmentWorkFollowupInheritance.Member(member.batchKey(),member.customerId()); }
}