perf(firstleg): batch appointment source and execution facts

2026-10-15 19:08:18 +08:00
parent 2631bd3af4
commit abf4e45e39
10 changed files with 248 additions and 61 deletions
@@ -10,6 +10,10 @@ public interface AppointmentWorkSourceMapper {
        String destinationWarehouseName, LocalDateTime departedLocalAt, LocalDateTime sailedEvidenceLocalAt, int version, long factVersion, long deliveryVersion) { }
    record Cargo(long inboundId, long customerId, long orderId, String orderNo, String inboundCode,
        String destinationKind, String destinationCode, String destinationName, int packageCount) { }
+    record ContainerCargo(long containerBookingId,long inboundId,long customerId,long orderId,String orderNo,String inboundCode,
+        String destinationKind,String destinationCode,String destinationName,int packageCount) {
+        public Cargo cargo() { return new Cargo(inboundId,customerId,orderId,orderNo,inboundCode,destinationKind,destinationCode,destinationName,packageCount); }
+    }

    String SOURCES = """
        SELECT b.id,b.cabinet_no,b.container_no,b.departure_port_name,b.destination_warehouse_name,
@@ -29,15 +33,19 @@ public interface AppointmentWorkSourceMapper {
    List<Source> sources();

    @Select("""
-        SELECT i.inbound_id, i.account_id customer_id, i.order_id, i.order_no,i.inbound_code,
+        <script>
+        SELECT p.container_booking_id,i.inbound_id, i.account_id customer_id, i.order_id, i.order_no,i.inbound_code,
          o.destination_type destination_kind,i.destination_warehouse_code destination_code,
          i.destination_warehouse_name destination_name,SUM(i.package_count) package_count
        FROM fl_loading_plan p JOIN fl_loading_plan_item i ON i.loading_plan_id=p.id AND i.deleted=0
          JOIN fl_inbound_order o ON o.id=i.inbound_id
-        WHERE p.container_booking_id=#{id} AND p.plan_status='COMPLETED'
-        GROUP BY i.inbound_id,i.account_id,i.order_id,i.order_no,i.inbound_code,o.destination_type,
+        WHERE p.container_booking_id IN
+          <foreach collection='ids' item='id' open='(' separator=',' close=')'>#{id}</foreach>
+          AND p.plan_status='COMPLETED'
+        GROUP BY p.container_booking_id,i.inbound_id,i.account_id,i.order_id,i.order_no,i.inbound_code,o.destination_type,
          i.destination_warehouse_code,i.destination_warehouse_name
-        ORDER BY i.inbound_id
+        ORDER BY p.container_booking_id,i.inbound_id
+        </script>
        """)
-    List<Cargo> cargo(@Param("id") long id);
+    List<ContainerCargo> cargo(@Param("ids") List<Long> ids);
 }
@@ -13,8 +13,8 @@ import java.util.*;
@Service
 public class AppointmentWorkSourceService {
    public interface OriginalFacts {
-        String establishedFlow(long containerBookingId);
-        Fact read(FactRequest request, ContainerSource source);
+        Map<Long,String> establishedFlows(List<Long> containerBookingIds);
+        List<Fact> read(List<FactRequest> requests, Map<Long,ContainerSource> sources);
    }
    private final AppointmentWorkSourceMapper mapper;
    private final OriginalFacts original;
@@ -37,17 +37,23 @@ public class AppointmentWorkSourceService {
        var page=rows.stream().filter(row->row.id()>cursor).limit(size+1L).toList();
        boolean complete=page.size()<=size;
        var selected=page.stream().limit(size).toList();
-        return new SourcePage(selected.stream().map(this::source).toList(),complete?null:scope+":"+selected.get(selected.size()-1).id(),complete,scope);
+        return new SourcePage(sources(selected),complete?null:scope+":"+selected.get(selected.size()-1).id(),complete,scope);
    }
    public ContainerSource container(long id) {
-        return container(id,mapper.sources());
-    }
-    private ContainerSource container(long id,List<AppointmentWorkSourceMapper.Source> rows) {
-        return rows.stream().filter(row->row.id()==id).findFirst().map(this::source)
+        var row=mapper.sources().stream().filter(candidate->candidate.id()==id).findFirst()
            .orElseThrow(()->new ResponseStatusException(HttpStatus.NOT_FOUND,"未找到有效已开船柜次"));
+        return sources(List.of(row)).get(0);
    }
-    private ContainerSource source(AppointmentWorkSourceMapper.Source row) {
-        var cargo=mapper.cargo(row.id());
+    private List<ContainerSource> sources(List<AppointmentWorkSourceMapper.Source> rows) {
+        if(rows.isEmpty()) return List.of();
+        var ids=rows.stream().map(AppointmentWorkSourceMapper.Source::id).toList();
+        var cargo=new HashMap<Long,List<AppointmentWorkSourceMapper.Cargo>>();
+        for(var item:mapper.cargo(ids)) cargo.computeIfAbsent(item.containerBookingId(),ignored->new ArrayList<>()).add(item.cargo());
+        var flows=original.establishedFlows(ids);
+        return rows.stream().map(row->source(row,cargo.getOrDefault(row.id(),List.of()),flows.get(row.id()))).toList();
+    }
+    private ContainerSource source(AppointmentWorkSourceMapper.Source row,List<AppointmentWorkSourceMapper.Cargo> cargo,String flow) {
+        if(flow==null) throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"柜次路线事实读取不完整");
        var requirements=cargo.stream().map(c->new Requirement("INBOUND:"+c.inboundId(),null,null,
            c.destinationKind(),row.departurePortName(),c.destinationName(),null,
            List.of(new Reference("CUSTOMER",Long.toString(c.customerId()),null,"所属客户"),
@@ -55,20 +61,22 @@ public class AppointmentWorkSourceService {
                new Reference("INBOUND",Long.toString(c.inboundId()),c.inboundCode(),"入仓单"),
                new Reference("CONTAINER",Long.toString(row.id()),row.cabinetNo(),"柜次")),List.of(),List.of())).toList();
        return new ContainerSource(Long.toString(row.id()),row.containerNo(),null,row.departedLocalAt(),row.sailedEvidenceLocalAt(),
-            hash(List.of(row,cargo)),original.establishedFlow(row.id()),null,requirements);
+            hash(List.of(row,cargo)),flow,null,requirements);
    }
    public List<Fact> facts(List<FactRequest> requests) {
        if(requests==null || requests.isEmpty() || requests.size()>200) throw bad("事实请求须包含1至200项");
-        Set<String> keys=new HashSet<>(); Map<Long,ContainerSource> sources=new HashMap<>();
-        // A fact batch shares one authoritative manifest, instead of scanning every sailed container per item.
-        var rows=mapper.sources();
-        return requests.stream().map(request->{
+        Set<String> keys=new HashSet<>(); var ids=new LinkedHashSet<Long>();
+        for(var request:requests) {
            if(request==null || request.workItemKey()==null || request.workItemKey().isBlank() || !keys.add(request.workItemKey())
                || !("PICKUP".equals(request.type()) || "EMPTY_RETURN".equals(request.type()) || "DELIVERY".equals(request.type()) && "DIRECT".equals(request.flow())))
                throw bad("事项身份或事实类型无效");
-            long id=id(request.containerBookingId());
-            return original.read(request,sources.computeIfAbsent(id,key->container(key,rows)));
-        }).toList();
+            ids.add(id(request.containerBookingId()));
+        }
+        var selected=mapper.sources().stream().filter(row->ids.contains(row.id())).toList();
+        if(selected.size()!=ids.size()) throw new ResponseStatusException(HttpStatus.NOT_FOUND,"未找到有效已开船柜次");
+        var sources=new LinkedHashMap<Long,ContainerSource>();
+        sources(selected).forEach(source->sources.put(Long.parseLong(source.containerBookingId()),source));
+        return original.read(requests,sources);
    }
    public String hash(Object value) {
        try {return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(json.writeValueAsBytes(value)));}
@@ -10,6 +10,41 @@ public interface DirectDeliveryMapper {
        int version,LocalDateTime actualAt,String timezone,String receivedBy,String evidenceNo,String remark,
        String idempotencyKey,String requestHash) { }
    record Container(long id,String recordStatus,LocalDateTime warehouseUnloadedAt) { }
+    record FlowEvidence(long id,LocalDateTime warehouseUnloadedAt,long evidenceCount) { }
+    record ContainerDelivery(long containerBookingId,long inboundId,String inboundCode,String orderNo,String destinationName,int packageCount,
+        int version,LocalDateTime actualAt,String timezone,String receivedBy,String evidenceNo,String remark,
+        String idempotencyKey,String requestHash) {
+        public Delivery delivery(){return new Delivery(inboundId,inboundCode,orderNo,destinationName,packageCount,version,
+            actualAt,timezone,receivedBy,evidenceNo,remark,idempotencyKey,requestHash);}
+    }
+    @Select("""
+        <script>
+        SELECT c.id,c.unloading_completed_at warehouse_unloaded_at,COUNT(d.inbound_id) evidence_count
+        FROM fl_container_booking c LEFT JOIN fl_direct_delivery_unloading d ON d.container_booking_id=c.id
+        WHERE c.id IN <foreach collection="ids" item="id" open="(" separator="," close=")">#{id}</foreach>
+        GROUP BY c.id,c.unloading_completed_at
+        </script>
+        """)
+    List<FlowEvidence> flowEvidence(@Param("ids") List<Long> ids);
+    @Select("""
+        <script>SELECT id,record_status,unloading_completed_at warehouse_unloaded_at FROM fl_container_booking
+        WHERE id IN <foreach collection="ids" item="id" open="(" separator="," close=")">#{id}</foreach></script>
+        """)
+    List<Container> containers(@Param("ids") List<Long> ids);
+    @Select("""
+        <script>
+        SELECT p.container_booking_id,i.inbound_id,MAX(i.inbound_code) inbound_code,MAX(i.order_no) order_no,
+          MAX(i.destination_warehouse_name) destination_name,SUM(i.package_count) package_count,
+          COALESCE(d.version,0) version,d.actual_at,d.timezone,d.received_by,d.evidence_no,d.remark,d.idempotency_key,d.request_hash
+        FROM fl_loading_plan p JOIN fl_loading_plan_item i ON i.loading_plan_id=p.id AND i.deleted=0
+        LEFT JOIN fl_direct_delivery_unloading d ON d.container_booking_id=p.container_booking_id AND d.inbound_id=i.inbound_id
+        WHERE p.container_booking_id IN <foreach collection="ids" item="id" open="(" separator="," close=")">#{id}</foreach>
+          AND p.plan_status='COMPLETED'
+        GROUP BY p.container_booking_id,i.inbound_id,d.version,d.actual_at,d.timezone,d.received_by,d.evidence_no,d.remark,d.idempotency_key,d.request_hash
+        ORDER BY p.container_booking_id,i.inbound_id
+        </script>
+        """)
+    List<ContainerDelivery> deliveriesBatch(@Param("ids") List<Long> ids);
    @Select("SELECT id,record_status,unloading_completed_at warehouse_unloaded_at FROM fl_container_booking WHERE id=#{id} FOR UPDATE")
    @Options(flushCache=Options.FlushCachePolicy.TRUE,useCache=false)
    Container lockContainer(@Param("id") long id);
@@ -114,6 +114,16 @@ public interface TerminalDrayageMapper extends BaseMapper<TerminalDrayageEntity>
         ORDER BY created_at DESC,id DESC
        """)
    List<History> histories(@Param("containerBookingId") long containerBookingId);
+    record PickupChanges(long containerBookingId,long schedules,long fleets) { }
+    @Select("""
+        <script>SELECT container_booking_id,
+          SUM(CASE WHEN action='SCHEDULE_PICKUP' THEN 1 ELSE 0 END) schedules,
+          SUM(CASE WHEN action='SELECT_FLEET' THEN 1 ELSE 0 END) fleets
+        FROM fl_terminal_drayage_history WHERE container_booking_id IN
+        <foreach collection="ids" item="id" open="(" separator="," close=")">#{id}</foreach>
+        GROUP BY container_booking_id</script>
+        """)
+    List<PickupChanges> pickupChanges(@Param("ids") List<Long> ids);

    @Insert("""
        INSERT INTO fl_terminal_drayage_history(id,drayage_id,container_booking_id,action,from_status,to_status,
@@ -3,6 +3,8 @@ package com.baiyue.firstleg.terminaldrayage.service;
 import com.baiyue.firstleg.portschedule.dto.AppointmentWorkSourceDtos.*;
 import com.baiyue.firstleg.portschedule.service.AppointmentWorkSourceService.OriginalFacts;
 import com.baiyue.firstleg.portschedule.service.CarrierScheduleQueryService;
+import com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ContainerActualFacts;
+import com.baiyue.firstleg.terminaldrayage.entity.TerminalDrayageEntity;
 import com.baiyue.firstleg.terminaldrayage.mapper.DirectDeliveryMapper;
 import com.baiyue.firstleg.terminaldrayage.mapper.TerminalDrayageMapper;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -26,17 +28,43 @@ public class AppointmentWorkTerminalFacts implements OriginalFacts {
        CarrierScheduleQueryService nodes,UnloadingEvidenceGuard unloading,ObjectMapper json) {
        this.terminal=terminal;this.direct=direct;this.nodes=nodes;this.unloading=unloading;this.json=json;
    }
-    public String establishedFlow(long id) {
-        var container=direct.container(id);
-        boolean transfer=container!=null && container.warehouseUnloadedAt()!=null;
-        boolean delivery=direct.evidenceCount(id)>0;
-        if(transfer && delivery) throw new ResponseStatusException(HttpStatus.CONFLICT,"货柜存在冲突的中转和直送执行事实");
-        return transfer?"TRANSFER":delivery?"DIRECT":"UNKNOWN";
+    public Map<Long,String> establishedFlows(List<Long> ids) {
+        var result=new HashMap<Long,String>();
+        if(ids.isEmpty()) return result;
+        for(var row:direct.flowEvidence(ids)) {
+            boolean transfer=row.warehouseUnloadedAt()!=null;
+            boolean delivery=row.evidenceCount()>0;
+            if(transfer && delivery) throw new ResponseStatusException(HttpStatus.CONFLICT,"货柜存在冲突的中转和直送执行事实");
+            result.put(row.id(),transfer?"TRANSFER":delivery?"DIRECT":"UNKNOWN");
+        }
+        return result;
    }
-    public Fact read(FactRequest request,ContainerSource source) {
-        long id=Long.parseLong(request.containerBookingId());
-        var node=nodes.containerActualFacts(List.of(id)).stream().findFirst().orElse(null);
-        var plan=terminal.find(id);
+    public List<Fact> read(List<FactRequest> requests,Map<Long,ContainerSource> sources) {
+        var ids=requests.stream().map(r->Long.parseLong(r.containerBookingId())).distinct().toList();
+        if(ids.isEmpty()) return List.of();
+        var actuals=new HashMap<Long,ContainerActualFacts>();
+        for(var row:nodes.containerActualFacts(ids)) actuals.put(row.containerBookingId(),row);
+        var plans=new HashMap<Long,TerminalDrayageEntity>();
+        for(var row:terminal.findAll(ids)) plans.put(row.getContainerBookingId(),row);
+        var pickupIds=requests.stream().filter(r->"PICKUP".equals(r.type())).map(r->Long.parseLong(r.containerBookingId())).distinct().toList();
+        var changes=new HashMap<Long,TerminalDrayageMapper.PickupChanges>();
+        if(!pickupIds.isEmpty()) for(var row:terminal.pickupChanges(pickupIds)) changes.put(row.containerBookingId(),row);
+        var deliveryIds=requests.stream().filter(r->"DELIVERY".equals(r.type())).map(r->Long.parseLong(r.containerBookingId())).distinct().toList();
+        var deliveries=new HashMap<Long,List<DirectDeliveryMapper.Delivery>>();
+        if(!deliveryIds.isEmpty()) for(var row:direct.deliveriesBatch(deliveryIds))
+            deliveries.computeIfAbsent(row.containerBookingId(),ignored->new ArrayList<>()).add(row.delivery());
+        var returnIds=requests.stream().filter(r->"EMPTY_RETURN".equals(r.type())).map(r->Long.parseLong(r.containerBookingId())).distinct()
+            .filter(id->(actuals.get(id)==null || actuals.get(id).emptyReturnedAt()==null)
+                && (plans.get(id)==null || plans.get(id).getActualEmptyReturnAt()==null)).toList();
+        var completed=returnIds.isEmpty()?Map.<Long,LocalDateTime>of():unloading.completedUnloading(returnIds);
+        return requests.stream().map(request->{
+            long id=Long.parseLong(request.containerBookingId());
+            return read(request,sources.get(id),actuals.get(id),plans.get(id),changes.get(id),
+                deliveries.getOrDefault(id,List.of()),completed.get(id));
+        }).toList();
+    }
+    private Fact read(FactRequest request,ContainerSource source,ContainerActualFacts node,TerminalDrayageEntity plan,
+        TerminalDrayageMapper.PickupChanges changes,List<DirectDeliveryMapper.Delivery> deliveries,LocalDateTime completedUnloading) {
        var references=new ArrayList<Reference>();
        source.requirements().forEach(r->r.references().forEach(ref->{if(!references.contains(ref))references.add(ref);}));
        var blocks=new ArrayList<String>(); var missing=new ArrayList<String>();
@@ -53,9 +81,8 @@ public class AppointmentWorkTerminalFacts implements OriginalFacts {
                if(plan!=null && plan.getFleetSupplierId()!=null)
                    responsibility=new Responsibility("SUPPLIER",plan.getFleetSupplierId().toString(),plan.getFleetSupplierName());
                // The initial arrangement/assignment establishes a baseline. Only later changes invalidate dependants.
-                var history=terminal.histories(id);
-                long reschedules=Math.max(0,history.stream().filter(h->"SCHEDULE_PICKUP".equals(h.action())).count()-1);
-                long reassignments=Math.max(0,history.stream().filter(h->"SELECT_FLEET".equals(h.action())).count()-1);
+                long reschedules=Math.max(0,(changes==null?0:changes.schedules())-1);
+                long reassignments=Math.max(0,(changes==null?0:changes.fleets())-1);
                impact=reschedules+":"+reassignments;
                if(at==null) {
                    if(node==null || node.arrivedDestinationPortAt()==null) blocks.add("实际到港后才能安排提柜");
@@ -67,7 +94,7 @@ public class AppointmentWorkTerminalFacts implements OriginalFacts {
            case "EMPTY_RETURN" -> {
                at=node==null?null:node.emptyReturnedAt();
                if(at==null && plan!=null) at=plan.getActualEmptyReturnAt();
-                if(at==null && unloading.completedUnloading(id)==null) blocks.add("海外仓卸柜或全部直送目的仓实际卸货后才能还空");
+                if(at==null && completedUnloading==null) blocks.add("海外仓卸柜或全部直送目的仓实际卸货后才能还空");
                else if(at==null) {code="RECORD_EXECUTION";label="确认实际还空";permission="platform:firstleg:terminal-drayage:empty-return:confirm";}
                // Return responsibility belongs to the work handoff, not the historical pickup carrier.
            }
@@ -75,7 +102,7 @@ public class AppointmentWorkTerminalFacts implements OriginalFacts {
                var requirement=source.requirements().stream().filter(r->Objects.equals(r.requirementKey(),request.requirementKey())).findFirst()
                    .orElseThrow(()->new ResponseStatusException(HttpStatus.CONFLICT,"直送需求与原柜来源不一致"));
                references.clear(); references.addAll(requirement.references());
-                var row=direct.deliveries(id).stream().filter(d->("INBOUND:"+d.inboundId()).equals(request.requirementKey())).findFirst()
+                var row=deliveries.stream().filter(d->("INBOUND:"+d.inboundId()).equals(request.requirementKey())).findFirst()
                    .orElseThrow(()->new ResponseStatusException(HttpStatus.CONFLICT,"直送需求与实际装柜来源不一致"));
                at=row.actualAt();zone=row.timezone();businessNo=row.inboundCode();
                if(at==null && (node==null || node.pickedUpAt()==null)) blocks.add("实际提柜后才能记录直送卸货");
@@ -6,6 +6,7 @@ import com.baiyue.common.core.api.ResultCode;
 import com.baiyue.common.core.exception.BizException;
 import org.springframework.stereotype.Service;
 import java.time.*;
+import java.util.*;

@Service
 public class UnloadingEvidenceGuard implements EmptyReturnPrerequisite {
@@ -16,11 +17,32 @@ public class UnloadingEvidenceGuard implements EmptyReturnPrerequisite {
        public LocalDateTime deadline(int hours){return instant().plus(Duration.ofHours(hours)).atZone(ZoneId.of(timezone)).toLocalDateTime();}
    }
    public DirectCompletion directCompletion(long id) {
-        var rows=mapper.deliveries(id);
+        return directCompletion(mapper.deliveries(id));
+    }
+    private DirectCompletion directCompletion(List<DirectDeliveryMapper.Delivery> rows) {
        if(rows.isEmpty() || rows.stream().anyMatch(row->row.actualAt()==null)) return null;
        return rows.stream().map(row->new DirectCompletion(row.actualAt(),row.timezone()))
            .max(java.util.Comparator.comparing(DirectCompletion::instant)).orElse(null);
    }
+    public Map<Long,LocalDateTime> completedUnloading(List<Long> ids) {
+        var result=new HashMap<Long,LocalDateTime>();
+        if(ids.isEmpty()) return result;
+        var pending=new ArrayList<Long>();
+        for(var container:mapper.containers(ids)) {
+            if(!"ACTIVE".equals(container.recordStatus())) continue;
+            if(container.warehouseUnloadedAt()!=null) result.put(container.id(),container.warehouseUnloadedAt());
+            else pending.add(container.id());
+        }
+        if(!pending.isEmpty()) {
+            var deliveries=new HashMap<Long,List<DirectDeliveryMapper.Delivery>>();
+            for(var row:mapper.deliveriesBatch(pending)) deliveries.computeIfAbsent(row.containerBookingId(),ignored->new ArrayList<>()).add(row.delivery());
+            for(long id:pending) {
+                var completed=directCompletion(deliveries.getOrDefault(id,List.of()));
+                if(completed!=null) result.put(id,completed.localAt());
+            }
+        }
+        return result;
+    }
    public LocalDateTime completedUnloading(long id) {
        var container=mapper.container(id);
        if(container==null || !"ACTIVE".equals(container.recordStatus())) return null;
@@ -7,6 +7,8 @@ import org.junit.jupiter.api.Test;
 import org.springframework.web.server.ResponseStatusException;
 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import com.baiyue.firstleg.portschedule.dto.AppointmentWorkSourceDtos.*;
 import static org.mockito.Mockito.*;
@@ -21,7 +23,7 @@ class AppointmentWorkSourceServiceTest {
    }
    @Test void completeCursorManifestIncludesUnboundSourcesAndRejectsScopeDrift() {
        when(mapper.sources()).thenReturn(List.of(source(1,1),source(2,1)));
-        when(mapper.cargo(anyLong())).thenReturn(List.of()); when(facts.establishedFlow(anyLong())).thenReturn("UNKNOWN");
+        when(mapper.cargo(anyList())).thenReturn(List.of()); stubFlows("UNKNOWN");
        var first=service.discover(null,1);
        assertFalse(first.complete()); assertNotNull(first.nextCursor());
        assertNull(first.containers().get(0).departedAt());
@@ -33,9 +35,11 @@ class AppointmentWorkSourceServiceTest {
    }
    @Test void sourceKeepsCustomerAndPhysicalInboundLineageWithoutDisplayingIdsAsNumbers() {
        when(mapper.sources()).thenReturn(List.of(source(1,1)));
-        when(mapper.cargo(1)).thenReturn(List.of(new AppointmentWorkSourceMapper.Cargo(8,9,10,"ORDER-10","IN-8","PRIVATE","P","客户仓",12)));
-        when(facts.establishedFlow(1)).thenReturn("DIRECT");
+        when(mapper.cargo(List.of(1L))).thenReturn(List.of(new AppointmentWorkSourceMapper.ContainerCargo(1,8,9,10,"ORDER-10","IN-8","PRIVATE","P","客户仓",12)));
+        stubFlows("DIRECT");
        var source=service.container(1);var requirement=source.requirements().get(0);
+        assertEquals(service.hash(List.of(source(1,1),List.of(new AppointmentWorkSourceMapper.Cargo(8,9,10,"ORDER-10","IN-8","PRIVATE","P","客户仓",12)))),source.sourceVersion());
+        assertEquals(source,service.discover(null,200).containers().get(0));
        assertEquals("INBOUND:8",requirement.requirementKey());assertEquals("PRIVATE",requirement.destinationKind());
        assertTrue(requirement.references().stream().anyMatch(r->r.kind().equals("CUSTOMER") && r.key().equals("9") && r.businessNo()==null));
    }
@@ -44,31 +48,42 @@ class AppointmentWorkSourceServiceTest {
        assertEquals(401,assertThrows(ResponseStatusException.class,()->guarded.discover(null,null,20)).getStatusCode().value());
        verifyNoInteractions(mapper);
    }
+    @Test void missingFlowEvidenceIsNotSilentlyPresentedAsUnknown() {
+        when(mapper.sources()).thenReturn(List.of(source(1,1)));
+        assertEquals(503,assertThrows(ResponseStatusException.class,()->service.container(1)).getStatusCode().value());
+    }
    @Test void factBatchReadsManifestOnceAndPreservesEachContainerIdentity() {
        when(mapper.sources()).thenReturn(IntStream.rangeClosed(1,100).mapToObj(id->source(id,1)).toList());
-        when(mapper.cargo(anyLong())).thenReturn(List.of());
+        when(mapper.cargo(anyList())).thenReturn(List.of());
+        stubFlows("UNKNOWN");
        var fact=mock(Fact.class);
        when(facts.read(any(),any())).thenAnswer(invocation->{
-            FactRequest request=invocation.getArgument(0); ContainerSource source=invocation.getArgument(1);
-            assertEquals(request.containerBookingId(),source.containerBookingId());
-            return fact;
+            List<FactRequest> requests=invocation.getArgument(0); Map<Long,ContainerSource> sources=invocation.getArgument(1);
+            assertEquals(100,sources.size());
+            requests.forEach(request->assertEquals(request.containerBookingId(),sources.get(Long.parseLong(request.containerBookingId())).containerBookingId()));
+            return requests.stream().map(request->fact).toList();
        });
        var requests=IntStream.rangeClosed(1,100).boxed().flatMap(id->List.of(
            request("PICKUP-"+id,"PICKUP",id),request("RETURN-"+id,"EMPTY_RETURN",id)).stream()).toList();
        assertEquals(200,service.facts(requests).size());
        verify(mapper,times(1)).sources();
-        for(long id=1;id<=100;id++) verify(mapper,times(1)).cargo(id);
-        verify(facts,times(200)).read(any(),any());
+        verify(mapper,times(1)).cargo(IntStream.rangeClosed(1,100).mapToObj(Long::valueOf).toList());
+        verify(facts,times(1)).establishedFlows(anyList());
+        verify(facts,times(1)).read(any(),any());
    }
    @Test void nextFactBatchReadsFreshManifestAndRejectsRemovedContainer() {
        when(mapper.sources()).thenReturn(List.of(source(1,1)),List.of());
-        when(mapper.cargo(1)).thenReturn(List.of());
+        when(mapper.cargo(List.of(1L))).thenReturn(List.of());
+        stubFlows("UNKNOWN");
        var request=request("PICKUP-1","PICKUP",1);
        service.facts(List.of(request));
        assertEquals(404,assertThrows(ResponseStatusException.class,()->service.facts(List.of(request))).getStatusCode().value());
        verify(mapper,times(2)).sources();
        verify(facts,times(1)).read(any(),any());
    }
+    private void stubFlows(String flow) {
+        when(facts.establishedFlows(anyList())).thenAnswer(invocation->((List<Long>)invocation.getArgument(0)).stream().collect(Collectors.toMap(id->id,id->flow)));
+    }
    private FactRequest request(String key,String type,long id) {
        return new FactRequest(key,type,"TRANSIT",Long.toString(id),null,null,null,List.of(),List.of(),List.of());
    }
@@ -8,6 +8,8 @@ import org.h2.jdbcx.JdbcDataSource;
 import org.junit.jupiter.api.Test;
 import java.time.LocalDateTime;
 import java.util.UUID;
+import java.util.List;
+import com.baiyue.firstleg.portschedule.mapper.AppointmentWorkSourceMapper;
 import static org.junit.jupiter.api.Assertions.*;

 class DirectDeliveryMapperTest {
@@ -20,8 +22,19 @@ class DirectDeliveryMapperTest {
            s.execute("CREATE TABLE fl_direct_delivery_unloading(container_booking_id BIGINT,inbound_id BIGINT,actual_at TIMESTAMP,timezone VARCHAR,received_by VARCHAR,evidence_no VARCHAR,remark VARCHAR,version INT,idempotency_key VARCHAR,request_hash VARCHAR,created_by BIGINT,created_by_name VARCHAR,PRIMARY KEY(container_booking_id,inbound_id),UNIQUE(container_booking_id,idempotency_key))");
            s.execute("INSERT INTO fl_container_booking VALUES(1,'ACTIVE',NULL)");s.execute("INSERT INTO fl_loading_plan VALUES(10,1,'COMPLETED')");
            s.execute("INSERT INTO fl_loading_plan_item VALUES(10,20,'I20','O20','D',3,0),(10,20,'I20','O20','D',4,0),(10,21,'I21','O21','D',5,0),(10,22,'I22','O22','D',100,1)");
+            s.execute("INSERT INTO fl_container_booking VALUES(2,'ACTIVE','2026-01-02 12:00:00'),(3,'ACTIVE',NULL)");
+            s.execute("INSERT INTO fl_loading_plan VALUES(11,2,'COMPLETED'),(12,3,'DRAFT')");
+            s.execute("INSERT INTO fl_loading_plan_item VALUES(11,20,'I20','O20','D',2,0),(12,20,'I20','O20','D',100,0)");
+            s.execute("ALTER TABLE fl_loading_plan_item ADD account_id BIGINT DEFAULT 7");
+            s.execute("ALTER TABLE fl_loading_plan_item ADD order_id BIGINT DEFAULT 8");
+            s.execute("ALTER TABLE fl_loading_plan_item ADD destination_warehouse_code VARCHAR DEFAULT 'D'");
+            s.execute("CREATE TABLE fl_inbound_order(id BIGINT,destination_type VARCHAR)");
+            s.execute("INSERT INTO fl_inbound_order VALUES(20,'PRIVATE'),(21,'AMAZON'),(22,'PRIVATE')");
+            s.execute("CREATE TABLE fl_terminal_drayage_history(container_booking_id BIGINT,action VARCHAR)");
+            s.execute("INSERT INTO fl_terminal_drayage_history VALUES(1,'SELECT_FLEET'),(1,'SCHEDULE_PICKUP'),(1,'SCHEDULE_PICKUP'),(1,'OTHER'),(2,'SELECT_FLEET')");
        }
        var config=new MybatisConfiguration();config.setMapUnderscoreToCamelCase(true);config.setEnvironment(new Environment("test",new JdbcTransactionFactory(),ds));config.addMapper(DirectDeliveryMapper.class);
+        config.addMapper(AppointmentWorkSourceMapper.class);config.addMapper(TerminalDrayageMapper.class);
        try(var session=new MybatisSqlSessionFactoryBuilder().build(config).openSession()) {
            var mapper=session.getMapper(DirectDeliveryMapper.class);assertEquals(1,mapper.lockContainer(1).id());
            assertEquals(2,mapper.deliveries(1).size());assertEquals(7,mapper.deliveries(1).get(0).packageCount());
@@ -29,6 +42,18 @@ class DirectDeliveryMapperTest {
            mapper.insert(1,20,at,"America/Los_Angeles","签收人","POD20",null,"key","hash",7,"admin");
            assertEquals(1,mapper.evidenceCount(1));assertEquals(1,mapper.keyCount(1,"key"));
            assertEquals(at,mapper.deliveries(1).get(0).actualAt());assertNull(mapper.deliveries(1).get(1).actualAt());
+            var batch=mapper.deliveriesBatch(List.of(1L,2L,3L));
+            assertEquals(3,batch.size());assertEquals(mapper.deliveries(1),batch.stream().filter(r->r.containerBookingId()==1).map(DirectDeliveryMapper.ContainerDelivery::delivery).toList());
+            assertEquals(2,batch.get(2).packageCount());assertNull(batch.get(2).actualAt());
+            assertEquals(3,mapper.containers(List.of(1L,2L,3L)).size());
+            var flows=mapper.flowEvidence(List.of(1L,2L,3L));
+            assertEquals(1,flows.stream().filter(r->r.id()==1).findFirst().orElseThrow().evidenceCount());
+            assertNotNull(flows.stream().filter(r->r.id()==2).findFirst().orElseThrow().warehouseUnloadedAt());
+            var cargo=session.getMapper(AppointmentWorkSourceMapper.class).cargo(List.of(1L,2L,3L));
+            assertEquals(3,cargo.size());assertEquals(7,cargo.get(0).packageCount());assertEquals(1,cargo.get(0).containerBookingId());
+            assertEquals(2,cargo.get(2).containerBookingId());assertEquals(2,cargo.get(2).packageCount());
+            var changes=session.getMapper(TerminalDrayageMapper.class).pickupChanges(List.of(1L,2L,3L));
+            assertEquals(new TerminalDrayageMapper.PickupChanges(1,2,1),changes.stream().filter(r->r.containerBookingId()==1).findFirst().orElseThrow());
            assertThrows(RuntimeException.class,()->mapper.insert(1,20,at,"America/Los_Angeles","签收人","POD20",null,"another","hash",7,"admin"));
            session.rollback();
        }
@@ -5,11 +5,11 @@ import com.baiyue.firstleg.portschedule.dto.CarrierScheduleQueryDtos.ContainerAc
 import com.baiyue.firstleg.portschedule.service.CarrierScheduleQueryService;
 import com.baiyue.firstleg.terminaldrayage.mapper.*;
 import com.baiyue.firstleg.terminaldrayage.entity.TerminalDrayageEntity;
-import com.baiyue.firstleg.terminaldrayage.dto.TerminalDrayageDtos.History;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.Test;
 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.Map;
 import static org.mockito.Mockito.*;
 import static org.junit.jupiter.api.Assertions.*;

@@ -22,22 +22,43 @@ class AppointmentWorkTerminalFactsTest {
    private final LocalDateTime at=LocalDateTime.parse("2026-01-01T12:30:00");
    private ContainerSource source(){return new ContainerSource("1","C1",null,at,at,"v","UNKNOWN",null,List.of());}
    private FactRequest request(String type){return new FactRequest("W1",type,"DIRECT","1",null,null,null,List.of(),List.of(),List.of());}
-    private History history(String action){return new History(action,"A","B",1,2,null,7L,"admin",at);}
+    private Fact read(FactRequest request,ContainerSource source) {return adapter.read(List.of(request),Map.of(1L,source)).get(0);}
+    @Test void twoHundredRequestsReadEachFactSetOnceAndKeepMissingFactsVisible() {
+        var ids=java.util.stream.LongStream.rangeClosed(1,100).boxed().toList();
+        var sources=new java.util.HashMap<Long,ContainerSource>();
+        var requests=new java.util.ArrayList<FactRequest>();
+        for(long id:ids) {
+            sources.put(id,new ContainerSource(Long.toString(id),"C"+id,null,at,at,"v","UNKNOWN",null,List.of()));
+            requests.add(new FactRequest("P"+id,"PICKUP","TRANSIT",Long.toString(id),null,null,null,List.of(),List.of(),List.of()));
+            requests.add(new FactRequest("R"+id,"EMPTY_RETURN","TRANSIT",Long.toString(id),null,null,null,List.of(),List.of(),List.of()));
+        }
+        var results=adapter.read(requests,sources);
+        assertEquals(200,results.size());
+        for(int i=0;i<results.size();i++) {assertEquals(requests.get(i).workItemKey(),results.get(i).workItemKey());assertFalse(results.get(i).blockReasons().isEmpty());}
+        verify(nodes).containerActualFacts(ids);verify(terminal).findAll(ids);verify(terminal).pickupChanges(ids);
+        verify(guard).completedUnloading(ids);verifyNoMoreInteractions(nodes,terminal,guard);verifyNoInteractions(direct);
+    }
+    @Test void batchedFlowConflictsStillFailClosed() {
+        when(direct.flowEvidence(List.of(1L,2L))).thenReturn(List.of(new DirectDeliveryMapper.FlowEvidence(1,null,1),new DirectDeliveryMapper.FlowEvidence(2,at,1)));
+        assertEquals(409,assertThrows(org.springframework.web.server.ResponseStatusException.class,()->adapter.establishedFlows(List.of(1L,2L))).getStatusCode().value());
+    }
    @Test void initialPickupPlanRetainsLocalDateAndDoesNotInventZoneOrReview() {
-        var plan=new TerminalDrayageEntity();plan.setPickupAppointmentAt(at);plan.setFleetSupplierId(4L);plan.setFleetSupplierName("车队");
-        when(terminal.find(1)).thenReturn(plan);
-        when(terminal.histories(1)).thenReturn(List.of(history("SELECT_FLEET"),history("SCHEDULE_PICKUP")));
-        var fact=adapter.read(request("PICKUP"),source());
+        var plan=new TerminalDrayageEntity();plan.setContainerBookingId(1L);plan.setPickupAppointmentAt(at);plan.setFleetSupplierId(4L);plan.setFleetSupplierName("车队");
+        when(terminal.findAll(List.of(1L))).thenReturn(List.of(plan));
+        when(terminal.pickupChanges(List.of(1L))).thenReturn(List.of(new TerminalDrayageMapper.PickupChanges(1,1,1)));
+        var fact=read(request("PICKUP"),source());
        assertEquals("0:0",fact.impactToken());assertEquals(at.toLocalDate(),fact.confirmed().date());
        assertNull(fact.timezone());assertTrue(fact.missingFields().contains("SOURCE_TIMEZONE"));
-        when(terminal.histories(1)).thenReturn(List.of(history("SELECT_FLEET"),history("SCHEDULE_PICKUP"),history("SCHEDULE_PICKUP")));
-        assertEquals("1:0",adapter.read(request("PICKUP"),source()).impactToken());
+        var returnRequest=new FactRequest("W2","EMPTY_RETURN","DIRECT","1",null,null,null,List.of(),List.of(),List.of());
+        assertEquals(fact,adapter.read(List.of(request("PICKUP"),returnRequest),Map.of(1L,source())).get(0));
+        when(terminal.pickupChanges(List.of(1L))).thenReturn(List.of(new TerminalDrayageMapper.PickupChanges(1,2,1)));
+        assertEquals("1:0",read(request("PICKUP"),source()).impactToken());
    }
    @Test void emptyReturnActualIsCompleteEvenWhenOriginalTimezoneIsUnknownAndDoesNotCopyPickupResponsibility() {
-        var plan=new TerminalDrayageEntity();plan.setFleetSupplierId(4L);plan.setFleetSupplierName("提柜承运人");
-        when(terminal.find(1)).thenReturn(plan);
+        var plan=new TerminalDrayageEntity();plan.setContainerBookingId(1L);plan.setFleetSupplierId(4L);plan.setFleetSupplierName("提柜承运人");
+        when(terminal.findAll(List.of(1L))).thenReturn(List.of(plan));
        when(nodes.containerActualFacts(List.of(1L))).thenReturn(List.of(new ContainerActualFacts(1L,2L,3,"STOPPED","EMPTY_RETURNED",at,null,null,null,null,at.minusDays(1),at)));
-        var fact=adapter.read(request("EMPTY_RETURN"),source());
+        var fact=read(request("EMPTY_RETURN"),source());
        assertEquals("COMPLETED",fact.executionStatus());assertNull(fact.actual().occurredAt());assertEquals(at,fact.actual().localOccurredAt());
        assertNull(fact.responsibility());assertNull(fact.action());assertNull(fact.businessNo());
    }
@@ -47,9 +68,10 @@ class AppointmentWorkTerminalFactsTest {
        var source=new ContainerSource("1","C1",null,at,at,"v","DIRECT",null,List.of(
            new Requirement("INBOUND:2",null,null,"PRIVATE","港","A",null,own,List.of(),List.of()),
            new Requirement("INBOUND:3",null,null,"PRIVATE","港","B",null,other,List.of(),List.of())));
-        when(direct.deliveries(1)).thenReturn(List.of(new DirectDeliveryMapper.Delivery(2,"IN2","O2","A",4,1,at,"America/Los_Angeles","a","P2",null,"k","h")));
+        when(direct.deliveriesBatch(List.of(1L))).thenReturn(List.of(new DirectDeliveryMapper.ContainerDelivery(1,2,"IN2","O2","A",4,1,at,"America/Los_Angeles","a","P2",null,"k","h")));
        var request=new FactRequest("W1","DELIVERY","DIRECT","1","INBOUND:2",null,null,List.of(),List.of(),List.of());
-        var fact=adapter.read(request,source);
+        var fact=read(request,source);
        assertEquals(own,fact.references());assertEquals("IN2",fact.businessNo());assertEquals("COMPLETED",fact.executionStatus());
+        assertEquals(fact,adapter.read(List.of(request,new FactRequest("W2","PICKUP","DIRECT","1",null,null,null,List.of(),List.of(),List.of())),Map.of(1L,source)).get(0));
    }
 }
@@ -9,6 +9,21 @@ import static org.mockito.Mockito.*;
 import static org.junit.jupiter.api.Assertions.*;

 class UnloadingEvidenceGuardTest {
+    @Test void batchRetainsInactiveMissingPartialAndZonedCompletionRules() {
+        var mapper=mock(DirectDeliveryMapper.class);var guard=new UnloadingEvidenceGuard(mapper);
+        var at=LocalDateTime.parse("2026-10-10T12:00:00");
+        var ids=List.of(1L,2L,3L,4L,5L,6L);
+        when(mapper.containers(ids)).thenReturn(List.of(new DirectDeliveryMapper.Container(1,"ACTIVE",at),
+            new DirectDeliveryMapper.Container(2,"ACTIVE",null),new DirectDeliveryMapper.Container(3,"ACTIVE",null),
+            new DirectDeliveryMapper.Container(4,"CANCELLED",at),new DirectDeliveryMapper.Container(5,"ACTIVE",null)));
+        when(mapper.deliveriesBatch(List.of(2L,3L,5L))).thenReturn(List.of(
+            new DirectDeliveryMapper.ContainerDelivery(2,20,"I20","O20","D",1,1,at,"America/New_York",null,null,null,null,null),
+            new DirectDeliveryMapper.ContainerDelivery(2,21,"I21","O21","D",1,1,at.minusHours(2),"America/Los_Angeles",null,null,null,null,null),
+            new DirectDeliveryMapper.ContainerDelivery(3,30,"I30","O30","D",1,1,at,"America/New_York",null,null,null,null,null),
+            new DirectDeliveryMapper.ContainerDelivery(3,31,"I31","O31","D",1,0,null,null,null,null,null,null,null)));
+        assertEquals(java.util.Map.of(1L,at,2L,at.minusHours(2)),guard.completedUnloading(ids));
+        verify(mapper).containers(ids);verify(mapper).deliveriesBatch(List.of(2L,3L,5L));verifyNoMoreInteractions(mapper);
+    }
    @Test void requiresEveryPhysicalInboundSourceAndAcceptsLatestActualEvidence() {
        var mapper=mock(DirectDeliveryMapper.class);var guard=new UnloadingEvidenceGuard(mapper);
        var container=new DirectDeliveryMapper.Container(1,"ACTIVE",null);