perf(lastleg): batch live appointment work facts and filter candidates

This commit is contained in:
2026-10-15 19:10:57 +08:00
parent abf4e45e39
commit 097bec33b1
24 changed files with 441 additions and 40 deletions
@@ -16,6 +16,17 @@ public interface AppointmentCargoVerificationMapper extends BaseMapper<Appointme
@Select("SELECT * FROM ll_appointment_cargo_verification WHERE customer_id=#{customerId} AND source_key=#{sourceKey} ORDER BY version DESC LIMIT 1")
AppointmentCargoVerification latest(@Param("customerId") String customerId,@Param("sourceKey") String sourceKey);
@Select("""
<script>
SELECT ranked.* FROM (
SELECT v.*,ROW_NUMBER() OVER(PARTITION BY customer_id,source_key ORDER BY version DESC) work_row
FROM ll_appointment_cargo_verification v WHERE
<foreach collection='refs' item='ref' separator=' OR '>(customer_id=#{ref.customerId} AND source_key=#{ref.sourceKey})</foreach>
) ranked WHERE work_row=1
</script>
""")
java.util.List<AppointmentCargoVerification> latestBatch(@Param("refs") java.util.List<AppointmentCoverageService.SourceRef> refs);
@Select("SELECT * FROM ll_appointment_cargo_verification WHERE customer_id=#{customerId} AND source_key=#{sourceKey} AND command_key=#{key}")
AppointmentCargoVerification command(@Param("customerId") String customerId,@Param("sourceKey") String sourceKey,@Param("key") String key);
@@ -27,4 +38,15 @@ public interface AppointmentCargoVerificationMapper extends BaseMapper<Appointme
WHERE customer_id=#{customerId} AND source_key=#{sourceKey} AND active=1
""")
Allocation allocated(@Param("customerId") String customerId,@Param("sourceKey") String sourceKey);
record SourceAllocation(String customerId,String sourceKey,long packages,long unknownItems) { }
@Select("""
<script>
SELECT customer_id,source_key,COALESCE(SUM(package_count),0) packages,COUNT(*)-COUNT(package_count) unknown_items
FROM ll_inbound_fba_appointment_item WHERE active=1 AND (
<foreach collection='refs' item='ref' separator=' OR '>(customer_id=#{ref.customerId} AND source_key=#{ref.sourceKey})</foreach>
) GROUP BY customer_id,source_key
</script>
""")
java.util.List<SourceAllocation> allocatedBatch(@Param("refs") java.util.List<AppointmentCoverageService.SourceRef> refs);
}
@@ -47,10 +47,36 @@ public class AppointmentCargoVerificationService {
var source=sources.resolve(sourceKey,customerId,destinationKey);
return inspect(source);
}
public record Inspection(AppointmentCoverageService.SourceRef reference,Context context,ResponseStatusException error) {
public Context requireContext() { if(error!=null) throw error; return context; }
}
/** Shares source and warehouse reads within this call only; never caches mutable business facts across requests. */
public List<Inspection> inspectBatch(List<AppointmentCoverageService.SourceRef> refs) {
var result=new ArrayList<Inspection>();
for(int start=0;start<refs.size();start+=100) {
var batch=refs.subList(start,Math.min(start+100,refs.size()));
var resolved=sources.inspectBatch(batch);
var valid=resolved.stream().filter(r->r.error()==null).map(AppointmentSourceService.Resolution::reference).toList();
var latest=new HashMap<List<String>,AppointmentCargoVerification>();
var allocations=new HashMap<List<String>,AppointmentCargoVerificationMapper.Allocation>();
if(!valid.isEmpty()) {
mapper.latestBatch(valid).forEach(r->latest.put(List.of(r.getCustomerId(),r.getSourceKey()),r));
mapper.allocatedBatch(valid).forEach(r->allocations.put(List.of(r.customerId(),r.sourceKey()),new AppointmentCargoVerificationMapper.Allocation(r.packages(),r.unknownItems())));
}
for(var resolution:resolved) {
var key=List.of(resolution.reference().customerId(),resolution.reference().sourceKey());
result.add(new Inspection(resolution.reference(),resolution.error()==null?inspect(resolution.source(),latest.get(key),allocations.getOrDefault(key,new AppointmentCargoVerificationMapper.Allocation(0,0))):null,resolution.error()));
}
}
return List.copyOf(result);
}
/** Uses an already-authorized batched source resolution; never repeats warehouse HTTP lookups per row. */
public Context inspect(AppointmentSourceService.Source source) {
var row=mapper.latest(source.customerId(),source.sourceKey());
var allocation=mapper.allocated(source.customerId(),source.sourceKey());
return inspect(source,row,allocation);
}
private Context inspect(AppointmentSourceService.Source source,AppointmentCargoVerification row,AppointmentCargoVerificationMapper.Allocation allocation) {
boolean changed=row!=null && changed(row,source);
Integer total=row==null?source.packages():row.getPackageCount();
var reasons=new ArrayList<>(source.blockReasons());
@@ -60,12 +60,22 @@ public class AppointmentSourceService {
}
/** One catalog request and batched source reads, including mixed first-leg and external bookings. */
public List<Source> resolveBatch(List<AppointmentCoverageService.SourceRef> requested) {
return resolveBatch(requested,true).stream().map(Resolution::requireSource).toList();
}
/** Read-only inspection keeps each source's prerequisite failure separate; infrastructure errors still escape. */
public record Resolution(AppointmentCoverageService.SourceRef reference,Source source,ResponseStatusException error) {
public Source requireSource() { if(error!=null) throw error; return source; }
}
public List<Resolution> inspectBatch(List<AppointmentCoverageService.SourceRef> requested) {
return resolveBatch(requested,false);
}
private List<Resolution> resolveBatch(List<AppointmentCoverageService.SourceRef> requested,boolean strict) {
if(requested==null || requested.isEmpty() || requested.size()>100) throw bad("每批请选择1至100票真实货源");
var keys=new LinkedHashSet<String>(); var orderIds=new LinkedHashSet<Long>();
var bookingRefs=new java.util.ArrayList<AppointmentExternalSourceGateway.SourceRef>();
for(var ref:requested) {
if(ref==null || blank(ref.sourceKey()) || ref.sourceKey().length()>160 || blank(ref.customerId()) || ref.customerId().length()>128) throw bad("请选择完整的真实货源");
if(!keys.add(ref.sourceKey())) throw bad("同一来源货物不能重复选择");
if(!keys.add(ref.sourceKey()) && strict) throw bad("同一来源货物不能重复选择");
if(ref.sourceKey().startsWith("ORDER:")) orderIds.add(orderId(ref.sourceKey()));
else if(ref.sourceKey().startsWith("BOOKING:") && !blank(ref.sourceKey().substring(8))) bookingRefs.add(new AppointmentExternalSourceGateway.SourceRef(ref.customerId(),ref.sourceKey().substring(8)));
else throw bad("不支持的货源类型");
@@ -77,17 +87,28 @@ public class AppointmentSourceService {
if(booking.orderId()!=null) orderIds.add(booking.orderId());
}
var orderRows=orderIds.isEmpty()?List.<com.baiyue.lastleg.entity.LastlegOrder>of():orders.selectBatchIds(orderIds);
var directOrders=requested.stream().filter(ref->ref.sourceKey().startsWith("ORDER:")).map(ref->new AppointmentExternalSourceGateway.OrderRef(ref.customerId(),orderId(ref.sourceKey()))).toList();
if(!directOrders.isEmpty() && !external.resolveOrders(directOrders).isEmpty()) throw bad("该货物来自外接登记,请使用原登记来源继续预约");
var directOrders=requested.stream().filter(ref->ref.sourceKey().startsWith("ORDER:")).map(ref->new AppointmentExternalSourceGateway.OrderRef(ref.customerId(),orderId(ref.sourceKey()))).distinct().toList();
var externalOrders=directOrders.isEmpty()?Set.<AppointmentExternalSourceGateway.OrderRef>of():external.resolveOrders(directOrders).stream()
.map(s->new AppointmentExternalSourceGateway.OrderRef(s.customerId(),s.orderId())).collect(Collectors.toSet());
if(strict && !externalOrders.isEmpty()) throw bad("该货物来自外接登记,请使用原登记来源继续预约");
var orderMap=orderRows.stream().collect(Collectors.toMap(com.baiyue.lastleg.entity.LastlegOrder::getId,Function.identity()));
var transferRefs=orderRows.stream().filter(o->"WAREHOUSE_TRANSFER".equals(o.getOrderType())).map(o->new AppointmentCoverageMapper.OrderRef(o.getCustomerId(),o.getId())).toList();
var destinations=transferRefs.isEmpty()?Map.<Long,AppointmentCoverageMapper.Source>of():destinationFacts.sources(transferRefs).stream().collect(Collectors.toMap(AppointmentCoverageMapper.Source::orderId,Function.identity()));
var containerIds=orderRows.stream().filter(o->"FIRST_LEG".equals(o.getOrderType())).map(com.baiyue.lastleg.entity.LastlegOrder::getSourceContainerBookingId).filter(Objects::nonNull).distinct().toList();
var gates=containerIds.isEmpty()?Map.<String,Boolean>of():shipping.selectBatchIds(containerIds).stream().collect(Collectors.toMap(g->g.getContainerBookingId(),g->g.getNodeSequence()!=null && g.getNodeSequence()>=40));
var catalog=catalog();
var result=requested.stream().map(ref->resolve(ref.sourceKey(),ref.customerId(),ref.destinationKey(),orderMap,bookingMap,gates,catalog,destinations)).toList();
var result=requested.stream().map(ref->{
try {
if(ref.sourceKey().startsWith("ORDER:") && externalOrders.contains(new AppointmentExternalSourceGateway.OrderRef(ref.customerId(),orderId(ref.sourceKey()))))
throw bad("该货物来自外接登记,请使用原登记来源继续预约");
return new Resolution(ref,resolve(ref.sourceKey(),ref.customerId(),ref.destinationKey(),orderMap,bookingMap,gates,catalog,destinations),null);
} catch(ResponseStatusException error) {
if(strict || error.getStatusCode().value()>=500) throw error;
return new Resolution(ref,null,error);
}
}).toList();
var materialized=new java.util.HashSet<Long>();
for(var source:result) if(source.orderId()!=null && !materialized.add(source.orderId())) throw bad("同一货物存在重复来源,请保留原登记来源");
if(strict) for(var resolution:result) { var source=resolution.requireSource(); if(source.orderId()!=null && !materialized.add(source.orderId())) throw bad("同一货物存在重复来源,请保留原登记来源"); }
return result;
}
private Source resolve(String sourceKey,String customerId,String destinationKey,
@@ -29,6 +29,12 @@ public class AppointmentWorkCommitmentService {
var row=mapper.commitment(key);
return key+":"+(row==null?0:row.getImpactVersion());
}
public java.util.Map<String,String> impactTokens(java.util.List<String> keys) {
var tokens=new java.util.HashMap<String,String>();
keys.forEach(key->tokens.put(key,key+":0"));
if(!keys.isEmpty()) mapper.commitments(keys).forEach(row->tokens.put(row.getEntityKey(),row.getEntityKey()+":"+row.getImpactVersion()));
return tokens;
}
public static boolean impacts(Commitment old,Commitment now) {
if(old.cancelled()!=now.cancelled()) return true;
boolean firstSameDayConfirmation=old.confirmed()==null && now.confirmed()!=null && old.target()!=null && old.target().date().equals(now.confirmed().date());
@@ -44,6 +44,8 @@ public interface AppointmentWorkMapper {
}
@Select("SELECT * FROM ll_appointment_work_commitment WHERE entity_key=#{key} FOR UPDATE") CommitmentRow lockCommitment(String key);
@Select("SELECT * FROM ll_appointment_work_commitment WHERE entity_key=#{key}") CommitmentRow commitment(String key);
@Select("<script>SELECT * FROM ll_appointment_work_commitment WHERE entity_key IN <foreach collection='keys' item='key' open='(' separator=',' close=')'>#{key}</foreach></script>")
List<CommitmentRow> commitments(@Param("keys") List<String> keys);
@Insert("INSERT INTO ll_appointment_work_commitment(entity_key,state_json,impact_version) VALUES(#{key},#{json},0)")
int insertCommitment(@Param("key") String key,@Param("json") String json);
@Update("UPDATE ll_appointment_work_commitment SET state_json=#{json},impact_version=#{version} WHERE entity_key=#{key}")
@@ -7,6 +7,7 @@ import java.time.*;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkFactGateway.*;
@@ -23,16 +24,29 @@ public class AppointmentWorkQueryService {
AppointmentWorkQueryService(AppointmentWorkStore store,AppointmentWorkFactGateway facts,Clock clock) { this.store=store; this.facts=facts; this.clock=clock; }
record Snapshot(List<AppointmentWorkState> states,Map<String,Fact> facts,List<WorkItem> items) { }
Snapshot snapshot(Actor actor) {
return snapshot(actor,s->true,true);
}
private Snapshot snapshot(Actor actor,Predicate<AppointmentWorkState> selected,boolean includeNext) {
String scopeVersion=store.requireInitialized();
if(!Objects.equals(scopeVersion,facts.discover(null,1).scopeVersion())) throw conflict("已开船工作范围变化,请先完成服务端同步");
var all=store.all();
var states=all.stream().filter(s->!s.isRetired()).toList();
var stateByKey=all.stream().collect(Collectors.toMap(AppointmentWorkState::getWorkItemKey,Function.identity()));
var selectedKeys=all.stream().filter(s->!s.isRetired() && selected.test(s)).map(AppointmentWorkState::getWorkItemKey).collect(Collectors.toSet());
var needed=new HashSet<>(selectedKeys);
if(includeNext) {
var pending=new ArrayDeque<>(selectedKeys);
while(!pending.isEmpty()) {
var state=stateByKey.get(pending.remove());
for(var key:effectiveKeys(state.getPrerequisites(),stateByKey)) if(stateByKey.containsKey(key) && needed.add(key)) pending.add(key);
}
}
var states=all.stream().filter(s->!s.isRetired() && needed.contains(s.getWorkItemKey())).toList();
var current=facts.facts(states);
var visible=states.stream().filter(s->canRead(s,current.get(s.getWorkItemKey()),actor)).toList();
var items=visible.stream().map(s->project(s,current.get(s.getWorkItemKey()),actor,clock)).toList();
var byKey=items.stream().collect(Collectors.toMap(WorkItem::workItemKey,Function.identity()));
var stateByKey=all.stream().collect(Collectors.toMap(AppointmentWorkState::getWorkItemKey,Function.identity()));
return new Snapshot(all,current,items.stream().map(i->withNext(i,next(i,byKey,stateByKey,new HashSet<>()))).toList());
return new Snapshot(all,current,items.stream().filter(i->selectedKeys.contains(i.workItemKey()))
.map(i->includeNext?withNext(i,next(i,byKey,stateByKey,new HashSet<>())):i).toList());
}
private Action next(WorkItem item,Map<String,WorkItem> items,Map<String,AppointmentWorkState> states,Set<String> visited) {
if(!visited.add(item.workItemKey()) || item.blockReasons().isEmpty() || item.reviewRequired()) return item.nextAction();
@@ -46,13 +60,13 @@ public class AppointmentWorkQueryService {
return item.nextAction();
}
public Summary summary(Actor actor) {
var items=snapshot(actor).items().stream().filter(i->i.executionStatus()!=ExecutionStatus.COMPLETED).toList();
var items=snapshot(actor,s->true,false).items().stream().filter(i->i.executionStatus()!=ExecutionStatus.COMPLETED).toList();
return new Summary(items.size(),items.stream().filter(i->followToday(i,clock)).count(),items.stream().filter(i->!i.overdueReasons().isEmpty()).count(),items.stream().filter(i->i.calendarDate()==null).count(),items.stream().filter(WorkItem::reviewRequired).count());
}
public CalendarResult calendar(LocalDate from,LocalDate to,Flow flow,Set<Type> types,Actor actor) {
if(from==null || to==null || to.isBefore(from) || ChronoUnit.DAYS.between(from,to)>=62) throw bad("日历范围须为1至62天");
if(types!=null && types.contains(Type.ROUTE)) throw bad("路线待确认需求请从全局待办查看");
var items=snapshot(actor).items().stream().filter(i->i.type()!=Type.ROUTE && i.calendarDate()!=null && !i.calendarDate().isBefore(from) && !i.calendarDate().isAfter(to))
var items=snapshot(actor,s->s.getType()!=Type.ROUTE && matches(s,flow,types,null),false).items().stream().filter(i->i.type()!=Type.ROUTE && i.calendarDate()!=null && !i.calendarDate().isBefore(from) && !i.calendarDate().isAfter(to))
.filter(i->flow==null || i.flow()==flow).filter(i->types==null || types.contains(i.type())).toList();
var groups=items.stream().collect(Collectors.groupingBy(WorkItem::calendarDate,TreeMap::new,Collectors.toList()));
var days=groups.entrySet().stream().map(e->new DayAggregate(e.getKey(),e.getValue().size(),e.getValue().stream().collect(Collectors.groupingBy(WorkItem::type,()->new EnumMap<>(Type.class),Collectors.counting())),counts(e.getValue()))).toList();
@@ -61,7 +75,7 @@ public class AppointmentWorkQueryService {
public PageResult<WorkItem> items(View view,int page,int size,LocalDate date,Flow flow,Set<Type> types,String containerId,Bucket bucket,String keyword,Actor actor) {
if(page<1 || size<1 || size>100 || keyword!=null && keyword.length()>128) throw bad("分页或搜索条件无效");
var word=keyword==null?"":keyword.trim().toLowerCase(Locale.ROOT);
var items=snapshot(actor).items().stream()
var items=snapshot(actor,s->matches(s,flow,types,containerId) && candidate(s,view),true).items().stream()
.filter(i->date!=null || i.executionStatus()!=ExecutionStatus.COMPLETED)
.filter(i->date==null || date.equals(i.calendarDate()))
.filter(i->flow==null || flow==i.flow()).filter(i->types==null || types.contains(i.type()))
@@ -89,6 +103,21 @@ public class AppointmentWorkQueryService {
requirementOwned(state,fact)?new RequirementDetails(state.isBookingRequired(),Objects.toString(state.getReceivingRequirements(),""),state.getRequirementReason()):null,state.getReturnHandoff());
}
WorkItem current(AppointmentWorkState state,Fact fact,Actor actor) { return project(state,fact,actor,clock); }
private boolean matches(AppointmentWorkState state,Flow flow,Set<Type> types,String containerId) {
return (flow==null || state.getFlow()==flow) && (types==null || types.contains(state.getType()))
&& (containerId==null || (state.getContainers().isEmpty()?containerId.equals(state.getContainerBookingId()):state.getContainers().stream().anyMatch(c->containerId.equals(c.containerBookingId()))));
}
private boolean candidate(AppointmentWorkState state,View view) {
// Follow-up dates and handoffs belong to the work state. Use the latest possible local day,
// not its stored timezone: the authoritative fact can supply a different current zone.
// Exact local-day, completion and permission checks still run after the live fact read.
return switch(view) {
case FOLLOW_TODAY -> pendingHandoff(state) || state.getFollowupDate()!=null
&& !state.getFollowupDate().isAfter(LocalDate.now(clock.withZone(ZoneOffset.MAX)));
case REVIEW -> state.isReviewRequired();
default -> true;
};
}
private String searchable(WorkItem i) {
return String.join(" ",Objects.toString(i.businessNo(),""),Objects.toString(i.batchNo(),""),Objects.toString(i.originName(),""),Objects.toString(i.destinationName(),""),i.containers().stream().map(c->Objects.toString(c.containerNo(),"")+" "+Objects.toString(c.bookingNo(),"")).collect(Collectors.joining(" "))).toLowerCase(Locale.ROOT);
}
@@ -145,6 +145,20 @@ public class InboundFbaAppointmentService {
}
return new FbaAppointmentBatchContext(snapshot,List.copyOf(allocations),List.copyOf(actions),reasons.stream().distinct().toList());
}
/** Original persisted appointment facts only; reading work status must not run split/merge eligibility. */
@Transactional(readOnly=true)
public List<FbaAppointmentResponse> batchSnapshots(List<String> keys) {
if(keys==null || keys.isEmpty()) return List.of();
if(keys.size()>200) throw bad("每次最多读取200个预约批次");
var ids=keys.stream().map(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("批次关联键无效");
}).distinct().toList();
var rows=appointmentMapper.selectBatchIds(ids);
if(rows.size()!=ids.size()) throw missing();
return responses(rows);
}
@Transactional(isolation=org.springframework.transaction.annotation.Isolation.READ_COMMITTED)
public FbaAppointmentResponse update(String operatorId,String appointmentNo,UpdateFbaAppointmentRequest request) {
if ("CANCELLED".equals(status(request.details().appointmentStatus()))) throw bad("取消预约请使用取消操作并填写实际取消原因");
@@ -16,4 +16,20 @@ public interface AppointmentWorkDeliveryMapper {
record Delivery(String customerId,String sourceKey,String destinationKey,Integer allocatedPackages,String dispatchStatus,String deliveryNo,String status,Integer packages,Integer signedPackages,LocalDateTime signedAt) { }
@Select("SELECT b.customer_id,b.source_key,b.destination_key,b.package_count allocated_packages,d.status dispatch_status,x.delivery_no,x.status,x.package_count packages,x.signed_packages,x.signed_at FROM ll_dispatch_appointment_allocation b JOIN ll_dispatch_order d ON d.id=b.dispatch_id AND d.status<>'CANCELLED' LEFT JOIN ll_delivery_order x ON x.dispatch_order_id=b.dispatch_id AND x.customer_id=b.customer_id AND CONCAT('FBA:',x.destination_warehouse_code)=b.destination_key WHERE b.batch_key=#{batch} ORDER BY b.dispatch_item_id,b.source_key,x.id LIMIT 1001")
List<Delivery> batch(String batch);
record BatchDelivery(String batchKey,String customerId,String sourceKey,String destinationKey,Integer allocatedPackages,String dispatchStatus,String deliveryNo,String status,Integer packages,Integer signedPackages,LocalDateTime signedAt) {
public Delivery delivery() { return new Delivery(customerId,sourceKey,destinationKey,allocatedPackages,dispatchStatus,deliveryNo,status,packages,signedPackages,signedAt); }
}
@Select("""
<script>
SELECT batch_key,customer_id,source_key,destination_key,allocated_packages,dispatch_status,delivery_no,status,packages,signed_packages,signed_at FROM (
SELECT b.batch_key,b.customer_id,b.source_key,b.destination_key,b.package_count allocated_packages,d.status dispatch_status,
x.delivery_no,x.status,x.package_count packages,x.signed_packages,x.signed_at,
ROW_NUMBER() OVER(PARTITION BY b.batch_key ORDER BY b.dispatch_item_id,b.source_key,x.id) work_row
FROM ll_dispatch_appointment_allocation b JOIN ll_dispatch_order d ON d.id=b.dispatch_id AND d.status&lt;&gt;'CANCELLED'
LEFT JOIN ll_delivery_order x ON x.dispatch_order_id=b.dispatch_id AND x.customer_id=b.customer_id AND CONCAT('FBA:',x.destination_warehouse_code)=b.destination_key
WHERE b.batch_key IN <foreach collection='batches' item='batch' open='(' separator=',' close=')'>#{batch}</foreach>
) ranked WHERE work_row &lt;= 1001 ORDER BY batch_key,work_row
</script>
""")
List<BatchDelivery> batches(@Param("batches") List<String> batches);
}
@@ -3,6 +3,7 @@ package com.baiyue.lastleg.inboundplan.work;
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.security.OverseasPermissions;
import org.springframework.stereotype.Component;
import java.util.*;
@@ -22,14 +23,43 @@ public class AppointmentWorkDeliveryProvider implements AppointmentWorkFactGatew
this.appointments=appointments; this.cargo=cargo; this.commitments=commitments; this.deliveries=deliveries; this.sourceMapping=sourceMapping;
}
public boolean supports(Type type,Flow flow) { return type==Type.DELIVERY && flow==Flow.TRANSFER; }
public List<Fact> read(List<FactRequest> requests) { return requests.stream().map(this::read).toList(); }
private Fact read(FactRequest request) {
if(request.batchKey()==null) return unbatched(request);
public List<Fact> read(List<FactRequest> requests) {
var mapped=new HashMap<String,List<AppointmentWorkSourceMapper.Source>>();
var privateSources=new HashMap<String,List<AppointmentWorkSourceMapper.Source>>();
var containers=requests.stream().filter(r->r.batchKey()==null && r.requirementKey()!=null && r.requirementKey().startsWith("INBOUND:")).map(FactRequest::containerBookingId).distinct().toList();
if(!containers.isEmpty()) sourceMapping.containerSourcesBatch(containers).forEach(s->mapped.computeIfAbsent(s.containerBookingId(),ignored->new ArrayList<>()).add(s.source()));
for(var request:requests) if(request.batchKey()==null && request.requirementKey()!=null && request.requirementKey().startsWith("INBOUND:")) {
var sources=mapped.getOrDefault(request.containerBookingId(),List.of());
if(sources.size()>1000) throw unavailable("柜次尾程来源超过完整读取上限");
var related=sources.stream().filter(s->request.requirementKey().equals("INBOUND:"+s.sourceInboundId())).toList();
if(!related.isEmpty() && related.stream().allMatch(s->"PRIVATE".equals(s.destinationType()))) privateSources.put(request.workItemKey(),related);
}
var refs=requests.stream().filter(r->r.batchKey()==null && !privateSources.containsKey(r.workItemKey())).flatMap(r->r.sources().stream())
.map(s->new AppointmentCoverageService.SourceRef(s.customerId(),s.sourceKey(),s.destinationKey())).distinct().toList();
var inspections=new HashMap<AppointmentCoverageService.SourceRef,AppointmentCargoVerificationService.Inspection>();
if(!refs.isEmpty()) cargo.inspectBatch(refs).forEach(i->inspections.put(i.reference(),i));
var batches=requests.stream().map(FactRequest::batchKey).filter(k->k!=null && !k.startsWith("DISPATCH:")).distinct().toList();
var snapshots=new HashMap<String,FbaAppointmentResponse>();
var deliveryRows=new HashMap<String,List<AppointmentWorkDeliveryMapper.Delivery>>();
Map<String,String> impacts=Map.of();
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);
}
var context=new BatchFacts(snapshots,deliveryRows,impacts);
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 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 context=appointments.batchContext(request.batchKey()); var a=context.appointment();
var a=context.appointments().get(request.batchKey());
var impact=context.impacts().get(request.batchKey());
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=deliveries.batch(a.batchKey());
var rows=context.deliveries().getOrDefault(a.batchKey(),List.of());
if(rows.size()>1000) throw unavailable("批次派送事实超过完整读取上限");
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.
@@ -46,25 +76,21 @@ public class AppointmentWorkDeliveryProvider implements AppointmentWorkFactGatew
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));
var references=new ArrayList<BusinessReference>(); references.add(new BusinessReference("FBA_APPOINTMENT",a.batchKey(),a.appointmentNo(),"送仓预约"));
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),commitments.impactToken(a.batchKey()),true,
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);
}
private Fact unbatched(FactRequest request) {
if(request.requirementKey()!=null && request.requirementKey().startsWith("INBOUND:")) {
var mapped=sourceMapping.containerSources(request.containerBookingId());
if(mapped.size()>1000) throw unavailable("柜次尾程来源超过完整读取上限");
var related=mapped.stream().filter(s->request.requirementKey().equals("INBOUND:"+s.sourceInboundId())).toList();
if(!related.isEmpty() && related.stream().allMatch(s->"PRIVATE".equals(s.destinationType()))) return customer(request,related);
}
private Fact unbatched(FactRequest request,Map<AppointmentCoverageService.SourceRef,AppointmentCargoVerificationService.Inspection> inspections) {
var blocks=new ArrayList<String>(); var missing=new ArrayList<String>(); Action action=null;
var fingerprints=new ArrayList<String>();
if(request.sources().isEmpty()) { blocks.add("尾程货源尚未建立,请先核对整柜来源与入仓计划"); missing.add("LASTLEG_SOURCE"); }
else {
for(var source:request.sources()) {
try {
var context=cargo.inspect(source.sourceKey(),source.customerId(),source.destinationKey());
var inspection=inspections.get(new AppointmentCoverageService.SourceRef(source.customerId(),source.sourceKey(),source.destinationKey()));
if(inspection==null) throw unavailable("预约货源批量读取不完整");
var context=inspection.requireContext();
fingerprints.add(context.toString()); blocks.addAll(context.blockReasons());
} catch(org.springframework.web.server.ResponseStatusException error) {
if(!AppointmentWorkSourceAvailability.unresolved(error)) throw error;
@@ -7,8 +7,32 @@ import java.util.List;
@Mapper
public interface AppointmentWorkReceivingMapper {
@Select("SELECT DISTINCT p.* FROM ll_inbound_plan p JOIN ll_inbound_plan_line l ON l.plan_id=p.id JOIN ll_order o ON o.id=l.order_id AND o.customer_id=l.customer_id WHERE o.source_container_booking_id=#{container} ORDER BY p.id LIMIT 1001")
List<InboundPlan> plans(String container);
@Select("SELECT * FROM ll_receiving_container_session WHERE source_container_booking_id=#{container} ORDER BY created_at DESC,id DESC LIMIT 1")
ReceivingContainerSession session(String container);
class ContainerPlan extends InboundPlan {
private String sourceContainerBookingId;
public String getSourceContainerBookingId() { return sourceContainerBookingId; }
public void setSourceContainerBookingId(String value) { sourceContainerBookingId=value; }
}
@Select("""
<script>
SELECT ranked.* FROM (
SELECT source.*,ROW_NUMBER() OVER(PARTITION BY source_container_booking_id ORDER BY id) work_row FROM (
SELECT DISTINCT p.*,o.source_container_booking_id FROM ll_inbound_plan p
JOIN ll_inbound_plan_line l ON l.plan_id=p.id JOIN ll_order o ON o.id=l.order_id AND o.customer_id=l.customer_id
WHERE o.source_container_booking_id IN
<foreach collection='containers' item='container' open='(' separator=',' close=')'>#{container}</foreach>
) source
) ranked WHERE work_row &lt;= 1001 ORDER BY source_container_booking_id,id
</script>
""")
List<ContainerPlan> plans(@Param("containers") List<String> containers);
@Select("""
<script>
SELECT ranked.* FROM (
SELECT s.*,ROW_NUMBER() OVER(PARTITION BY source_container_booking_id ORDER BY created_at DESC,id DESC) work_row
FROM ll_receiving_container_session s WHERE source_container_booking_id IN
<foreach collection='containers' item='container' open='(' separator=',' close=')'>#{container}</foreach>
) ranked WHERE work_row=1
</script>
""")
List<ReceivingContainerSession> sessions(@Param("containers") List<String> containers);
}
@@ -4,6 +4,7 @@ import com.baiyue.lastleg.appointmentwork.AppointmentWorkFactGateway;
import com.baiyue.lastleg.security.OverseasPermissions;
import org.springframework.stereotype.Component;
import java.util.*;
import java.util.stream.Collectors;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkFactGateway.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkModels.*;
import static com.baiyue.lastleg.appointmentwork.AppointmentWorkPolicy.*;
@@ -13,11 +14,15 @@ public class AppointmentWorkReceivingProvider implements AppointmentWorkFactGate
private final AppointmentWorkReceivingMapper mapper;
public AppointmentWorkReceivingProvider(AppointmentWorkReceivingMapper mapper) { this.mapper=mapper; }
public boolean supports(Type type,Flow flow) { return type==Type.WAREHOUSE_RECEIPT && flow==Flow.TRANSFER; }
public List<Fact> read(List<FactRequest> requests) { return requests.stream().map(this::read).toList(); }
private Fact read(FactRequest request) {
var plans=mapper.plans(request.containerBookingId());
public List<Fact> read(List<FactRequest> requests) {
if(requests.isEmpty()) return List.of();
var ids=requests.stream().map(FactRequest::containerBookingId).distinct().toList();
var plans=mapper.plans(ids).stream().collect(Collectors.groupingBy(AppointmentWorkReceivingMapper.ContainerPlan::getSourceContainerBookingId));
var sessions=mapper.sessions(ids).stream().collect(Collectors.toMap(s->s.getSourceContainerBookingId(),s->s));
return requests.stream().map(r->read(r,plans.getOrDefault(r.containerBookingId(),List.of()),sessions.get(r.containerBookingId()))).toList();
}
private Fact read(FactRequest request,List<AppointmentWorkReceivingMapper.ContainerPlan> plans,com.baiyue.lastleg.inboundplan.pda.entity.ReceivingContainerSession session) {
if(plans.size()>1000) throw unavailable("柜次接仓计划超过完整读取上限");
var session=mapper.session(request.containerBookingId());
var dates=plans.stream().map(p->p.getPlannedArrivalDate()).filter(Objects::nonNull).distinct().sorted().toList();
var target=dates.isEmpty()?null:new Slot(dates.get(0),null,null,null,null);
boolean complete=session!=null && session.getCompletedAt()!=null && Set.of("COMPLETED","COMPLETED_WITH_DIFFERENCE").contains(session.getStatus());
@@ -8,6 +8,20 @@ public interface AppointmentWorkSourceMapper {
record Source(Long id,String customerId,String sourceOrderId,String sourceInboundId,String destinationType,String fbaWarehouseCode,Integer packageCount) { }
@Select("SELECT id,customer_id,source_order_id,source_inbound_id,destination_type,fba_warehouse_code,package_count FROM ll_order WHERE source_container_booking_id=#{container} ORDER BY id LIMIT 1001")
List<Source> containerSources(String container);
record ContainerSource(String containerBookingId,Long id,String customerId,String sourceOrderId,String sourceInboundId,String destinationType,String fbaWarehouseCode,Integer packageCount) {
public Source source() { return new Source(id,customerId,sourceOrderId,sourceInboundId,destinationType,fbaWarehouseCode,packageCount); }
}
@Select("""
<script>
SELECT container_booking_id,id,customer_id,source_order_id,source_inbound_id,destination_type,fba_warehouse_code,package_count FROM (
SELECT source_container_booking_id container_booking_id,id,customer_id,source_order_id,source_inbound_id,destination_type,fba_warehouse_code,package_count,
ROW_NUMBER() OVER(PARTITION BY source_container_booking_id ORDER BY id) work_row
FROM ll_order WHERE source_container_booking_id IN
<foreach collection='containers' item='container' open='(' separator=',' close=')'>#{container}</foreach>
) ranked WHERE work_row &lt;= 1001 ORDER BY container_booking_id,id
</script>
""")
List<ContainerSource> containerSourcesBatch(@Param("containers") List<String> containers);
record DispatchSource(long dispatchId,String dispatchNo,String status,long orderId,String customerId,String sourceKey,String destinationKey,long packages) { }
@Select("SELECT b.dispatch_id,d.dispatch_no,d.status,o.id order_id,b.customer_id,b.source_key,b.destination_key,SUM(b.package_count) packages FROM ll_dispatch_appointment_allocation b JOIN ll_dispatch_order d ON d.id=b.dispatch_id JOIN ll_order o ON b.destination_key=CONCAT('CUSTOMER:',o.id) AND b.customer_id=o.customer_id WHERE b.batch_key=CONCAT('DISPATCH:',d.id) AND o.source_container_booking_id=#{container} GROUP BY b.dispatch_id,d.dispatch_no,d.status,o.id,b.customer_id,b.source_key,b.destination_key ORDER BY b.dispatch_id,o.id LIMIT 1001")
List<DispatchSource> privateDispatches(String container);