package com.baiyue.lastleg.delivery; import com.baiyue.lastleg.delivery.mapper.*; import com.baiyue.lastleg.delivery.model.SourceDeliveryCompletionPolicy.*; import com.baiyue.lastleg.delivery.model.*; import com.baiyue.lastleg.delivery.entity.*; import com.baiyue.lastleg.delivery.dto.DeliveryDocumentDtos.*; import com.baiyue.lastleg.delivery.service.*; import com.baiyue.lastleg.delivery.gateway.DeliveryDocumentFileGateway; import com.baiyue.lastleg.customervisit.service.CustomerVisitService; import com.baiyue.lastleg.entity.*; import com.baiyue.lastleg.mapper.*; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.ibatis.annotations.Select; import org.junit.jupiter.api.Test; import org.springframework.jdbc.core.*; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; import org.springframework.jdbc.datasource.*; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.test.util.ReflectionTestUtils; import java.lang.reflect.Proxy; import java.time.LocalDateTime; import java.util.*; import java.util.concurrent.*; import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; /** Production document/query/policy/outbox flow, with actual SQL state and transactions. * Document BaseMapper plumbing is implemented with JDBC answers; source annotation SQL is executed verbatim. */ class DeliverySourcePublicationPersistenceTest { static class Fixture { final DriverManagerDataSource ds = new DriverManagerDataSource("jdbc:h2:mem:"+UUID.randomUUID()+";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=5000","sa",""); final JdbcTemplate db=new JdbcTemplate(ds); final NamedParameterJdbcTemplate named=new NamedParameterJdbcTemplate(ds); final DataSourceTransactionManager manager=new DataSourceTransactionManager(ds); DeliveryDocumentService service=new DeliveryDocumentService(); volatile boolean failOutbox, failRevision; DeliverySourceEvidenceMapper mapper; volatile Runnable beforeFirstSourceRead = () -> {}; final Map> sourceLocks=new ConcurrentHashMap<>(); Fixture() { org.springframework.test.util.ReflectionTestUtils.setField(service,"fleetPool",mock(com.baiyue.lastleg.fleet.business.FleetPoolOrderBridge.class)); db.execute("CREATE TABLE ll_order(id BIGINT PRIMARY KEY,customer_id VARCHAR(20),source_region VARCHAR(20),source_inbound_id VARCHAR(30),package_count BIGINT,UNIQUE(customer_id,source_region,source_inbound_id))"); db.execute("CREATE TABLE ll_dispatch_order(id BIGINT,customer_id VARCHAR(20),depart_at TIMESTAMP)"); db.execute("CREATE TABLE ll_delivery_order(id BIGINT PRIMARY KEY,customer_id VARCHAR(20),dispatch_order_id BIGINT,delivery_group_key VARCHAR(100),package_count BIGINT,signed_packages BIGINT,signed_at TIMESTAMP,outbound_no VARCHAR(30))"); db.execute("CREATE TABLE ll_dispatch_pda_task(id BIGINT,customer_id VARCHAR(20),dispatch_order_id BIGINT)"); db.execute("CREATE TABLE ll_dispatch_pda_pick(id BIGINT,customer_id VARCHAR(20),task_id BIGINT,dispatch_item_id BIGINT)"); db.execute("CREATE TABLE ll_dispatch_pda_pick_order(id BIGINT,customer_id VARCHAR(20),pick_id BIGINT,order_id BIGINT,packages BIGINT,receipt_id BIGINT,return_receipt_id BIGINT)"); db.execute("CREATE TABLE ll_dispatch_item(id BIGINT,customer_id VARCHAR(20),dispatch_order_id BIGINT,destination_type VARCHAR(20),delivery_address VARCHAR(100),postal_code VARCHAR(20),destination_warehouse_code VARCHAR(30),fba_warehouse_code VARCHAR(30),destination_warehouse_id VARCHAR(30))"); db.execute("CREATE TABLE ll_exception_order(id BIGINT,customer_id VARCHAR(20),delivery_order_id BIGINT,exception_no VARCHAR(30),last_mile_order_id BIGINT)"); db.execute("CREATE TABLE ll_inventory_ledger(id BIGINT,customer_id VARCHAR(20),business_order_no VARCHAR(30),outbound_order_no VARCHAR(30),business_type VARCHAR(30),delta_packages BIGINT,occurred_at TIMESTAMP,source_event_id VARCHAR(128))"); db.execute("CREATE TABLE ll_return_stock_receipt(id BIGINT,customer_id VARCHAR(20),order_id BIGINT,inventory_ledger_id BIGINT,received_packages BIGINT,exception_order_id BIGINT,source_outbound_no VARCHAR(30),source_outbound_pallet_item_id BIGINT,source_inventory_position_id BIGINT)"); db.execute("ALTER TABLE ll_order ADD sub_order_status VARCHAR(128)"); db.execute("ALTER TABLE ll_delivery_order ADD version BIGINT DEFAULT 0"); db.execute("ALTER TABLE ll_delivery_order ADD status VARCHAR(32) DEFAULT 'DELIVERING'"); db.execute("ALTER TABLE ll_delivery_order ADD documents_json CLOB"); db.execute("CREATE TABLE proof_revision(delivery_id BIGINT,command_key VARCHAR(128),request_hash VARCHAR(128),snapshot CLOB, UNIQUE(delivery_id,command_key))"); db.execute("CREATE TABLE proof_outbox(event_key VARCHAR(255) PRIMARY KEY,aggregate_id VARCHAR(128),customer_id VARCHAR(128),payload CLOB)"); db.execute("CREATE TABLE proof_operation(delivery_id BIGINT,node VARCHAR(80),result CLOB)"); mapper = (DeliverySourceEvidenceMapper)Proxy.newProxyInstance(getClass().getClassLoader(),new Class[]{DeliverySourceEvidenceMapper.class},(proxy,method,args) -> { if(method.getName().equals("source")) beforeFirstSourceRead.run(); if(method.getName().equals("lockSource")) sourceLocks.computeIfAbsent(Thread.currentThread().getName(),k->new CopyOnWriteArrayList<>()).add((Long)args[1]); String sql=String.join(" ",method.getAnnotation(Select.class).value()).replaceAll("#\\{(\\w+)\\}",":$1"); String name=method.getName(); String parameter=name.equals("source")?"sourceInboundId":name.equals("dispatchIds")||name.equals("lockSource")?"orderId":"dispatchId"; var rows=named.query(sql,Map.of("customerId",args[0],parameter,args[1]),(r,n)-> switch(name) { case "source","lockSource" -> new Source(r.getString("source_inbound_id"),r.getLong("order_id"),r.getLong("packages")); case "dispatchIds" -> r.getLong(1); case "deliveries" -> new Delivery(r.getLong("delivery_id"),r.getLong("dispatch_id"),r.getString("group_key"),r.getLong("packages"),r.getObject("signed_packages",Long.class),r.getObject("signed_at",LocalDateTime.class),r.getObject("departed_at",LocalDateTime.class)); case "picks" -> new Pick(r.getLong("attribution_id"),r.getLong("dispatch_id"),r.getString("group_key"),r.getLong("order_id"),r.getLong("packages"),r.getObject("receipt_id",Long.class),r.getObject("return_receipt_id",Long.class),r.getObject("return_ledger_id",Long.class),r.getObject("return_packages",Long.class)); case "returns" -> new Returned(r.getLong("ledger_id"),r.getLong("delivery_id"),r.getLong("order_id"),r.getLong("packages"),r.getObject("received_at",LocalDateTime.class)); default -> throw new IllegalStateException(name); }); return name.equals("source")||name.equals("lockSource") ? rows.isEmpty()?null:rows.get(0) : rows; }); var orders=mock(DeliveryOrderMapper.class); when(orders.lockByNo(anyString(),anyString())).thenAnswer(a -> delivery(Long.parseLong(a.getArgument(1)),true)); when(orders.updateById(any(DeliveryOrder.class))).thenAnswer(a -> { DeliveryOrder d=a.getArgument(0); return db.update("UPDATE ll_delivery_order SET signed_packages=?,signed_at=?,version=?,status=?,documents_json=? WHERE id=?",d.getSignedPackages(),d.getSignedAt(),d.getVersion(),d.getStatus(),d.getDocumentsJson(),d.getId()); }); var tails=mock(LastlegOrderMapper.class); when(tails.selectList(any())).thenReturn(List.of()); // Current delivery relation deliberately unavailable. when(tails.selectBatchIds(anyCollection())).thenAnswer(a -> { Collection ids=a.getArgument(0); var result=new ArrayList(); for(Object id:ids) result.add(tail(((Number)id).longValue())); return result; }); when(tails.selectById(anyLong())).thenAnswer(a -> tail(a.getArgument(0))); when(tails.updateById(any(LastlegOrder.class))).thenAnswer(a -> { LastlegOrder t=a.getArgument(0); return db.update("UPDATE ll_order SET sub_order_status=? WHERE id=?",t.getSubOrderStatus(),t.getId()); }); var revisions=mock(DeliveryDocumentRevisionMapper.class); when(revisions.findCommand(anyString(),anyLong(),anyString())).thenAnswer(a -> { var rows=db.query("SELECT * FROM proof_revision WHERE delivery_id=? AND command_key=?",(r,n)-> { var v=new DeliveryDocumentRevision();v.setRequestHash(r.getString("request_hash"));v.setSnapshotJson(r.getString("snapshot"));return v; },a.getArgument(1),a.getArgument(2)); return rows.isEmpty()?null:rows.get(0); }); when(revisions.insert(any(DeliveryDocumentRevision.class))).thenAnswer(a->{ DeliveryDocumentRevision v=a.getArgument(0);if(failRevision)throw new IllegalStateException("Injected revision failure");return db.update("INSERT INTO proof_revision VALUES(?,?,?,?)",v.getDeliveryOrderId(),v.getIdempotencyKey(),v.getRequestHash(),v.getSnapshotJson()); }); var operations=mock(DeliveryOperationMapper.class); when(operations.returnReceipts(anyString(),anyLong())).thenAnswer(a -> db.query( "SELECT l.id,l.delta_packages,l.occurred_at FROM ll_inventory_ledger l JOIN ll_exception_order e ON e.exception_no=l.business_order_no WHERE e.delivery_order_id=?", (r,n)->new DeliveryReturnReceiptFact(r.getLong(1),1L,1L,r.getLong(2),r.getObject(3,LocalDateTime.class)),new Object[]{a.getArgument(1)})); when(operations.insert(any(DeliveryOperation.class))).thenAnswer(a->{DeliveryOperation o=a.getArgument(0);return db.update("INSERT INTO proof_operation VALUES(?,?,?)",o.getDeliveryOrderId(),o.getOperationNode(),o.getResult());}); var outbox=mock(LastlegEventOutboxMapper.class); when(outbox.insert(any(LastlegEventOutbox.class))).thenAnswer(a->{LastlegEventOutbox o=a.getArgument(0);if(failOutbox)throw new IllegalStateException("Injected outbox failure");return db.update("INSERT INTO proof_outbox VALUES(?,?,?,?)",o.getEventKey(),o.getAggregateId(),o.getCustomerId(),o.getPayloadJson());}); when(outbox.findByEventKey(anyString())).thenAnswer(a->{var rows=db.query("SELECT * FROM proof_outbox WHERE event_key=?",(r,n)->{ var o=new LastlegEventOutbox();o.setAggregateId(r.getString("aggregate_id"));o.setCustomerId(r.getString("customer_id"));return o;},new Object[]{a.getArgument(0)});return rows.isEmpty()?null:rows.get(0);}); var publisher=new DeliveryCompletedOutboxService(); var json=new ObjectMapper().findAndRegisterModules(); ReflectionTestUtils.setField(publisher,"outbox",outbox);ReflectionTestUtils.setField(publisher,"objectMapper",json); var outbound=mock(OutboundOrderMapper.class);var shipped=new OutboundOrder();shipped.setStatus("OUTBOUNDED"); when(outbound.findByNo(anyString(),anyString())).thenReturn(shipped); var files=mock(DeliveryDocumentFileGateway.class); when(files.metadata(anyString(),anyLong(),any(),anyLong())).thenReturn(new com.baiyue.lastleg.delivery.dto.DeliveryDocumentDtos.File(1L,"receipt.png","image/png",1,LocalDateTime.of(2026,1,1,0,0))); Map fields=Map.ofEntries(Map.entry("orders",orders),Map.entry("outboundOrders",outbound),Map.entry("revisions",revisions),Map.entry("operations",operations),Map.entry("files",files),Map.entry("objectMapper",json),Map.entry("lastlegOrders",tails),Map.entry("deliveryCompletedOutbox",publisher),Map.entry("customerVisits",mock(CustomerVisitService.class)),Map.entry("sourceEvidence",mapper),Map.entry("sourceCompletion",new SourceDeliveryCompletionQuery(mapper))); fields.forEach((name,value)->ReflectionTestUtils.setField(service,name,value)); var proxyFactory=new org.springframework.aop.framework.ProxyFactory(service);proxyFactory.setProxyTargetClass(true); proxyFactory.addAdvice(new org.springframework.transaction.interceptor.TransactionInterceptor(manager,new org.springframework.transaction.annotation.AnnotationTransactionAttributeSource())); service=(DeliveryDocumentService)proxyFactory.getProxy(); } DeliveryOrder delivery(long id,boolean lock) { return db.queryForObject("SELECT * FROM ll_delivery_order WHERE id=?"+(lock?" FOR UPDATE":""),(r,n)->{ var d=new DeliveryOrder();d.setId(id);d.setCustomerId("C");d.setDeliveryNo(String.valueOf(id));d.setOutboundNo(r.getString("outbound_no"));d.setDispatchOrderId(r.getLong("dispatch_order_id"));d.setDeliveryGroupKey(r.getString("delivery_group_key"));d.setPackageCount(r.getInt("package_count"));d.setSignedPackages(r.getObject("signed_packages",Integer.class));d.setSignedAt(r.getObject("signed_at",LocalDateTime.class));d.setVersion(r.getLong("version"));d.setStatus(r.getString("status"));d.setDocumentsJson(r.getString("documents_json"));return d; },id); } LastlegOrder tail(long id) { return db.queryForObject("SELECT * FROM ll_order WHERE id=?",(r,n)->{var o=new LastlegOrder();o.setId(id);o.setCustomerId("C");o.setSourceRegion("FIRSTLEG");o.setSourceInboundId(r.getString("source_inbound_id"));o.setPackageCount(r.getInt("package_count"));o.setSubOrderStatus(r.getString("sub_order_status"));o.setOrderNo("LM-"+id);o.setInboundCode("IN-"+id);o.setSourceOrderId("SOURCE-"+id);o.setSourceOrderNo("FL-"+id);o.setRootOrderId("ROOT-"+id);return o;},id); } TransactionTemplate tx() { var tx=new TransactionTemplate(manager);tx.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);return tx; } Save request(String key) { return new Save(0L,key,null,List.of(new Binding(DeliveryDocumentCategory.RECEIPT,List.of(1L)))); } void sign(long id,String key) { service.save("C",String.valueOf(id),"TEST",request(key)); } void source(long id,int packages) { db.update("INSERT INTO ll_order(id,customer_id,source_region,source_inbound_id,package_count) VALUES(?,'C','FIRSTLEG',?,?)",id,"S-"+id,packages); } void shipment(long id,int packages) { db.update("INSERT INTO ll_dispatch_order VALUES(?,'C','2026-01-01 01:00:00')",id); db.update("INSERT INTO ll_delivery_order(id,customer_id,dispatch_order_id,delivery_group_key,package_count,outbound_no) VALUES(?,'C',?,'CODE:A',?,?)",id,id,packages,"OUT-"+id); db.update("INSERT INTO ll_dispatch_pda_task VALUES(?,'C',?)",id,id); db.update("INSERT INTO ll_dispatch_item VALUES(?,'C',?,'FBA',NULL,NULL,'A',NULL,NULL)",id,id); db.update("INSERT INTO ll_dispatch_pda_pick VALUES(?,'C',?,?)",id,id,id); } void pick(long id,long dispatch,long order,int packages,Long receipt) { db.update("INSERT INTO ll_dispatch_pda_pick_order VALUES(?,'C',?,?,?, ?,?)",id,dispatch,order,packages,receipt==null?id:null,receipt); } void returned(long id,long delivery,long order,int packages,LocalDateTime at) { db.update("INSERT INTO ll_exception_order VALUES(?,'C',?,?,?)",id,delivery,"EX-"+id,order); db.update("INSERT INTO ll_inventory_ledger VALUES(?,'C',?,?,'RETURN_STOCK_IN',?,?,?)",id,"EX-"+id,"OUT-"+delivery,packages,at,"RETURN_SOURCE:7:8"); db.update("INSERT INTO ll_return_stock_receipt VALUES(?,'C',?,?,?,?,?,7,8)",id,order,id,packages,id,"OUT-"+delivery); } List published() { return db.queryForList("SELECT aggregate_id FROM proof_outbox ORDER BY aggregate_id",String.class); } } @Test void mixedDeliveryPublishesOnlyFullySignedSourceAndReplayDoesNotRepublish() { var f=new Fixture();f.source(101,4);f.source(102,6);f.shipment(1,10);f.pick(1,1,101,4,null);f.pick(2,1,102,6,null); f.returned(30,1,101,2,LocalDateTime.of(2026,1,2,0,0));f.sign(1,"same-key"); assertThat(f.published()).containsExactly("S-102");assertThat(f.tail(101).getSubOrderStatus()).contains("WAIT_DELIVERY"); assertThat(f.tail(102).getSubOrderStatus()).contains("DELIVERY:DELIVERED"); f.sign(1,"same-key");assertThat(f.published()).containsExactly("S-102"); assertThat(f.db.queryForObject("SELECT COUNT(*) FROM proof_revision",Integer.class)).isEqualTo(1); // Re-dispatch only the two previously unsigned boxes. Original current delivery pointers are absent. f.shipment(2,2);f.db.update("UPDATE ll_dispatch_order SET depart_at=? WHERE id=2",LocalDateTime.now().plusSeconds(1)); f.pick(3,2,101,2,30L); // Keep deterministic chronology without sleeping or depending on wall clock granularity. f.db.update("UPDATE ll_dispatch_order SET depart_at='2026-01-03 00:00:00' WHERE id=2"); f.sign(2,"redelivery");assertThat(f.published()).containsExactly("S-101","S-102"); } @Test void postSignatureRedeliveryDoesNotFillNeverShippedFourBoxes() { var f=new Fixture();f.source(101,10);f.shipment(1,6);f.pick(1,1,101,6,null);f.sign(1,"first"); f.db.update("UPDATE ll_delivery_order SET signed_at='2026-01-02 00:00:00' WHERE id=1"); f.returned(30,1,101,4,LocalDateTime.of(2026,1,3,0,0)); f.shipment(2,4);f.db.update("UPDATE ll_dispatch_order SET depart_at='2026-01-04 00:00:00' WHERE id=2"); f.pick(2,2,101,4,30L);f.sign(2,"repeat-four"); assertThat(f.published()).isEmpty();assertThat(f.tail(101).getSubOrderStatus()).contains("WAIT_DELIVERY"); } @Test void concurrentSignaturesPublishOnlyAfterLastSourceLockOwnerSeesBoth() throws Exception { var f=new Fixture();f.source(101,10);f.source(102,10);f.shipment(1,10);f.shipment(2,10); f.pick(1,1,102,5,null);f.pick(2,1,101,5,null); // B,A input f.pick(3,2,101,5,null);f.pick(4,2,102,5,null); // A,B input var firstLocked=new CountDownLatch(1);var secondUpdated=new CountDownLatch(1);var pool=Executors.newFixedThreadPool(2); f.beforeFirstSourceRead=()->{if(Thread.currentThread().getName().equals("second-signer"))secondUpdated.countDown();}; try { var first=pool.submit(()->f.tx().execute(s->{ Thread.currentThread().setName("first-signer"); f.delivery(1,true);f.mapper.lockSource("C",101);firstLocked.countDown();await(secondUpdated); var result=f.service.save("C","1","TEST",f.request("one"));assertThat(f.published()).isEmpty();return result; })); var second=pool.submit(()->{Thread.currentThread().setName("second-signer");await(firstLocked);return f.tx().execute(s->{ assertThat(f.db.queryForObject("SELECT COUNT(*) FROM ll_delivery_order WHERE signed_at IS NOT NULL",Integer.class)).isZero(); return f.service.save("C","2","TEST",f.request("two")); });}); first.get(10,TimeUnit.SECONDS);second.get(10,TimeUnit.SECONDS); assertThat(f.published()).containsExactly("S-101","S-102");f.sign(2,"two");assertThat(f.published()).containsExactly("S-101","S-102"); assertThat(f.sourceLocks.get("first-signer")).containsExactly(101L,101L,102L); assertThat(f.sourceLocks.get("second-signer")).containsExactly(101L,102L); } finally {pool.shutdownNow();} } @Test void springProxyAppliesReadCommittedAndFailureRollsBackSignatureSourceAndOutbox() { for(boolean revisionFailure:List.of(false,true)) { var f=new Fixture();f.source(101,4);f.shipment(1,4);f.pick(1,1,101,4,null); assertThat(org.springframework.aop.support.AopUtils.isAopProxy(f.service)).isTrue(); f.beforeFirstSourceRead=()->assertThat(org.springframework.transaction.support.TransactionSynchronizationManager.getCurrentTransactionIsolationLevel()) .isEqualTo(TransactionDefinition.ISOLATION_READ_COMMITTED); f.failRevision=revisionFailure;f.failOutbox=!revisionFailure; assertThatThrownBy(()->f.sign(1,"failure")).isInstanceOf(IllegalStateException.class).hasMessageContaining("Injected"); assertThat(f.delivery(1,false).getSignedAt()).isNull();assertThat(f.delivery(1,false).getSignedPackages()).isNull(); assertThat(f.delivery(1,false).getVersion()).isZero();assertThat(f.tail(101).getSubOrderStatus()).isNull(); assertThat(f.published()).isEmpty(); assertThat(f.db.queryForObject("SELECT COUNT(*) FROM proof_revision",Integer.class)).isZero(); assertThat(f.db.queryForObject("SELECT COUNT(*) FROM proof_operation",Integer.class)).isZero(); f.failRevision=false;f.failOutbox=false;f.sign(1,"failure");assertThat(f.published()).containsExactly("S-101"); } } @Test void multiHopPreSignatureRedeliveryPublishesWhenLastUnsignedBoxIsSigned() { var f=new Fixture();f.source(101,4);f.shipment(1,4);f.pick(1,1,101,4,null); f.returned(30,1,101,2,LocalDateTime.of(2026,1,2,0,0));f.sign(1,"first");assertThat(f.published()).isEmpty(); f.shipment(2,2);f.db.update("UPDATE ll_dispatch_order SET depart_at='2026-01-03 00:00:00' WHERE id=2");f.pick(2,2,101,2,30L); f.returned(40,2,101,1,LocalDateTime.of(2026,1,4,0,0));f.sign(2,"second");assertThat(f.published()).isEmpty(); f.shipment(3,1);f.db.update("UPDATE ll_dispatch_order SET depart_at='2026-01-05 00:00:00' WHERE id=3");f.pick(3,3,101,1,40L); f.sign(3,"third");assertThat(f.published()).containsExactly("S-101"); } @Test void unrelatedUnattributedDestinationDoesNotBlockSignedSource() { var f=new Fixture();f.source(101,4);f.shipment(1,4);f.pick(1,1,101,4,null); f.db.update("INSERT INTO ll_delivery_order(id,customer_id,dispatch_order_id,delivery_group_key,package_count,outbound_no) VALUES(2,'C',1,'CODE:B',6,'OUT-1')"); f.sign(1,"source-a");assertThat(f.published()).containsExactly("S-101"); assertThat(f.delivery(2,false).getSignedAt()).isNull(); } private static void await(CountDownLatch latch) { try {if(!latch.await(6,TimeUnit.SECONDS))throw new IllegalStateException("Timed out");} catch(InterruptedException e){Thread.currentThread().interrupt();throw new IllegalStateException(e);} } }