diff --git a/deploy/sql/头程数据库统一增量-MySQL8.sql b/deploy/sql/头程数据库统一增量-MySQL8.sql index 5a75105ef..d0ea8a737 100644 --- a/deploy/sql/头程数据库统一增量-MySQL8.sql +++ b/deploy/sql/头程数据库统一增量-MySQL8.sql @@ -5259,7 +5259,7 @@ CREATE TRIGGER trg_carrier_schedule_node_before_update BEFORE UPDATE ON fl_carri CREATE TRIGGER trg_carrier_schedule_query_before_insert BEFORE INSERT ON fl_carrier_schedule_query FOR EACH ROW BEGIN IF NOT EXISTS (SELECT 1 FROM fl_ocean_booking booking - WHERE booking.id=NEW.ocean_booking_id AND booking.customer_id=NEW.customer_id) THEN + WHERE booking.id=NEW.ocean_booking_id AND COALESCE(booking.customer_id,0)=NEW.customer_id) THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='船司查询客户与订舱不一致'; END IF; IF NEW.container_booking_id IS NOT NULL AND NOT EXISTS ( diff --git a/firstleg/src/main/java/com/baiyue/firstleg/outbound/dispatch/service/OutboundDispatchAutoConfirmService.java b/firstleg/src/main/java/com/baiyue/firstleg/outbound/dispatch/service/OutboundDispatchAutoConfirmService.java index 5880fdf02..9b89a4ae4 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/outbound/dispatch/service/OutboundDispatchAutoConfirmService.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/outbound/dispatch/service/OutboundDispatchAutoConfirmService.java @@ -1,10 +1,10 @@ package com.baiyue.firstleg.outbound.dispatch.service; -import com.baiyue.common.core.exception.BizException; import com.baiyue.firstleg.outbound.dispatch.domain.OutboundDispatchStatus; import com.baiyue.firstleg.outbound.dispatch.dto.OutboundDispatchDtos.ActionRequest; import jakarta.annotation.Resource; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; /** Real proxy-owned transaction boundary for repeated system convergence after customs completion. */ @@ -12,19 +12,13 @@ import org.springframework.transaction.annotation.Transactional; public class OutboundDispatchAutoConfirmService { @Resource private OutboundDispatchService dispatch; - @Transactional + @Transactional(propagation = Propagation.REQUIRES_NEW) public boolean confirmIfReady(long containerBookingId) { var current=dispatch.detail(containerBookingId); if(OutboundDispatchStatus.WAITING_FOR_LOADING.name().equals(current.aggregate().dispatchStatus()))return true; if(!current.aggregate().gate().passed())return false; - try{ - dispatch.confirm(containerBookingId,0L,"SYSTEM","CUSTOMS-AUTO-DISPATCH-"+containerBookingId, - new ActionRequest(current.aggregate().version(),"整柜报关完成且全部离仓门禁满足，系统自动确认离仓")); - return true; - }catch(BizException ex){ - var after=dispatch.detail(containerBookingId); - if(OutboundDispatchStatus.WAITING_FOR_LOADING.name().equals(after.aggregate().dispatchStatus()))return true; - throw ex; - } + dispatch.confirm(containerBookingId,0L,"SYSTEM","CUSTOMS-AUTO-DISPATCH-"+containerBookingId, + new ActionRequest(current.aggregate().version(),"整柜报关完成且全部离仓门禁满足，系统自动确认离仓")); + return true; } } diff --git a/firstleg/src/main/resources/db/phase3-carrier-schedule-query-schema.sql b/firstleg/src/main/resources/db/phase3-carrier-schedule-query-schema.sql index f8357e255..428238bd6 100644 --- a/firstleg/src/main/resources/db/phase3-carrier-schedule-query-schema.sql +++ b/firstleg/src/main/resources/db/phase3-carrier-schedule-query-schema.sql @@ -209,7 +209,7 @@ DELIMITER $$ CREATE TRIGGER trg_carrier_schedule_query_before_insert BEFORE INSERT ON fl_carrier_schedule_query FOR EACH ROW BEGIN IF NOT EXISTS (SELECT 1 FROM fl_ocean_booking booking - WHERE booking.id=NEW.ocean_booking_id AND booking.customer_id=NEW.customer_id) THEN + WHERE booking.id=NEW.ocean_booking_id AND COALESCE(booking.customer_id,0)=NEW.customer_id) THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='船司查询客户与订舱不一致'; END IF; IF NEW.container_booking_id IS NOT NULL AND NOT EXISTS ( diff --git a/firstleg/src/test/java/com/baiyue/firstleg/outbound/dispatch/OutboundDispatchAutoConfirmServiceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/outbound/dispatch/OutboundDispatchAutoConfirmServiceTest.java index 3e11cd5eb..73dde8142 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/outbound/dispatch/OutboundDispatchAutoConfirmServiceTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/outbound/dispatch/OutboundDispatchAutoConfirmServiceTest.java @@ -2,15 +2,33 @@ package com.baiyue.firstleg.outbound.dispatch; import com.baiyue.firstleg.outbound.dispatch.dto.OutboundDispatchDtos.*; import com.baiyue.firstleg.outbound.dispatch.service.*; +import com.baiyue.common.core.api.ResultCode; +import com.baiyue.common.core.exception.BizException; +import org.h2.jdbcx.JdbcDataSource; import org.junit.jupiter.api.*; import org.mockito.*; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource; +import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.interceptor.NameMatchTransactionAttributeSource; +import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; import java.time.LocalDateTime; import java.util.List; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.*; class OutboundDispatchAutoConfirmServiceTest { @@ -20,6 +38,58 @@ class OutboundDispatchAutoConfirmServiceTest { @Test void waitsWithoutWritingUntilEveryOtherGatePasses(){when(dispatch.detail(22L)).thenReturn(detail(false,"READY_FOR_DISPATCH",3));assertThat(service.confirmIfReady(22L)).isFalse();verify(dispatch,never()).confirm(anyLong(),anyLong(),anyString(),anyString(),any());} @Test void usesSystemIdentityAndStableIdempotencyKey(){when(dispatch.detail(22L)).thenReturn(detail(true,"READY_FOR_DISPATCH",3));assertThat(service.confirmIfReady(22L)).isTrue();verify(dispatch).confirm(eq(22L),eq(0L),eq("SYSTEM"),eq("CUSTOMS-AUTO-DISPATCH-22"),argThat(r->r.expectedVersion()==3));} @Test void concurrentReplayAlreadyConfirmedConvergesWithoutSecondMutation(){when(dispatch.detail(22L)).thenReturn(detail(true,"WAITING_FOR_LOADING",4));assertThat(service.confirmIfReady(22L)).isTrue();verify(dispatch,never()).confirm(anyLong(),anyLong(),anyString(),anyString(),any());} - @Test void proxyOwnedAutoBoundaryAndConfirmMutationAreTransactional()throws Exception{assertThat(OutboundDispatchAutoConfirmService.class.getMethod("confirmIfReady",long.class).getAnnotation(Transactional.class)).isNotNull();assertThat(OutboundDispatchService.class.getMethod("confirm",long.class,long.class,String.class,String.class,ActionRequest.class).getAnnotation(Transactional.class)).isNotNull();} + @Test void proxyOwnedAutoBoundaryAndConfirmMutationAreTransactional()throws Exception{assertThat(OutboundDispatchAutoConfirmService.class.getMethod("confirmIfReady",long.class).getAnnotation(Transactional.class).propagation()).isEqualTo(Propagation.REQUIRES_NEW);assertThat(OutboundDispatchService.class.getMethod("confirm",long.class,long.class,String.class,String.class,ActionRequest.class).getAnnotation(Transactional.class)).isNotNull();} + @Test void anUncommittedDepartureCannotHideCarrierTaskCreationFailure(){ + when(dispatch.detail(22L)).thenReturn(detail(true,"READY_FOR_DISPATCH",3), detail(true,"WAITING_FOR_LOADING",4)); + BizException failure=new BizException(ResultCode.CONFLICT,"船司查询客户与订舱不一致"); + when(dispatch.confirm(anyLong(),anyLong(),anyString(),anyString(),any())).thenThrow(failure); + assertThatThrownBy(()->service.confirmIfReady(22L)).isSameAs(failure); + verify(dispatch,times(1)).detail(22L); + } + + @Test void afterCommitFailureRollsBackDepartureAndTaskTogetherThenIndependentRetryCommitsBoth(){ + JdbcDataSource database=new JdbcDataSource(); + database.setURL("jdbc:h2:mem:auto-dispatch-"+UUID.randomUUID()+";DB_CLOSE_DELAY=-1"); + JdbcTemplate jdbc=new JdbcTemplate(database); + jdbc.execute("CREATE TABLE business_fact(kind VARCHAR(32) PRIMARY KEY)"); + var manager=new DataSourceTransactionManager(database); + AtomicBoolean rejectCarrier=new AtomicBoolean(true); + when(dispatch.detail(22L)).thenAnswer(invocation->detail(true, + jdbc.queryForObject("SELECT COUNT(*) FROM business_fact WHERE kind='OUTBOUND'",Integer.class)>0 + ? "WAITING_FOR_LOADING" : "READY_FOR_DISPATCH",3)); + when(dispatch.confirm(anyLong(),anyLong(),anyString(),anyString(),any())).thenAnswer(invocation->{ + jdbc.update("INSERT INTO business_fact(kind) VALUES('OUTBOUND')"); + if(rejectCarrier.get())throw new BizException(ResultCode.CONFLICT,"carrier task insert failed"); + jdbc.update("INSERT INTO business_fact(kind) VALUES('CARRIER_QUERY')"); + return detail(true,"WAITING_FOR_LOADING",4); + }); + var dispatchAttributes=new NameMatchTransactionAttributeSource(); + dispatchAttributes.addTransactionalMethod("confirm",new RuleBasedTransactionAttribute()); + ProxyFactory dispatchFactory=new ProxyFactory(dispatch); + dispatchFactory.setProxyTargetClass(true); + dispatchFactory.addAdvice(new TransactionInterceptor(manager,dispatchAttributes)); + ReflectionTestUtils.setField(service,"dispatch",dispatchFactory.getProxy()); + ProxyFactory autoFactory=new ProxyFactory(service); + autoFactory.setProxyTargetClass(true); + autoFactory.addAdvice(new TransactionInterceptor(manager,new AnnotationTransactionAttributeSource())); + var auto=(OutboundDispatchAutoConfirmService)autoFactory.getProxy(); + AtomicReference callbackFailure=new AtomicReference<>(); + new TransactionTemplate(manager).executeWithoutResult(status->{ + jdbc.update("INSERT INTO business_fact(kind) VALUES('CUSTOMS_COMPLETED')"); + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization(){ + @Override public void afterCommit(){ + try{auto.confirmIfReady(22L);}catch(RuntimeException failure){callbackFailure.set(failure);} + } + }); + }); + assertThat(callbackFailure.get()).isInstanceOf(BizException.class).hasMessage("carrier task insert failed"); + assertThat(jdbc.queryForList("SELECT kind FROM business_fact",String.class)).containsExactly("CUSTOMS_COMPLETED"); + rejectCarrier.set(false); + assertThat(auto.confirmIfReady(22L)).isTrue(); + assertThat(jdbc.queryForList("SELECT kind FROM business_fact",String.class)) + .containsExactlyInAnyOrder("CUSTOMS_COMPLETED","OUTBOUND","CARRIER_QUERY"); + assertThat(auto.confirmIfReady(22L)).isTrue(); + verify(dispatch,times(2)).confirm(anyLong(),anyLong(),anyString(),anyString(),any()); + } private static Detail detail(boolean gate,String status,int version){var result=new ContainerGateResult(gate,List.of(),LocalDateTime.now());var aggregate=new Aggregate(22L,"CAB","C",44L,"COMPLETED",66L,"RELEASED",77L,"DEPARTED",status,version,List.of(),result);return new Detail(aggregate,List.of());} } diff --git a/firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQuerySqlContractTest.java b/firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQuerySqlContractTest.java index c5b2d01d9..93ddbab9f 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQuerySqlContractTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/portschedule/repository/CarrierScheduleQuerySqlContractTest.java @@ -4,6 +4,8 @@ import com.baiyue.firstleg.portschedule.mapper.CarrierScheduleQueryMapper; import org.apache.ibatis.annotations.Select; import org.apache.ibatis.annotations.Update; import org.junit.jupiter.api.Test; +import org.h2.jdbcx.JdbcDataSource; +import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.core.io.ClassPathResource; import java.nio.charset.StandardCharsets; @@ -14,6 +16,34 @@ import java.util.List; import static org.assertj.core.api.Assertions.assertThat; class CarrierScheduleQuerySqlContractTest { + @Test + void deployedTriggerPredicateMapsPlatformBookingToScopeZeroAndStillRejectsOtherCustomers() throws Exception { + JdbcDataSource database = new JdbcDataSource(); + database.setURL("jdbc:h2:mem:carrier-scope-" + java.util.UUID.randomUUID() + ";DB_CLOSE_DELAY=-1"); + JdbcTemplate jdbc = new JdbcTemplate(database); + jdbc.execute("CREATE TABLE fl_ocean_booking(id BIGINT PRIMARY KEY,customer_id BIGINT)"); + jdbc.update("INSERT INTO fl_ocean_booking(id,customer_id) VALUES(1,NULL),(2,10001),(3,10002)"); + String unified = Files.readString(Path.of("..", "deploy", "sql", "头程数据库统一增量-MySQL8.sql"), + StandardCharsets.UTF_8); + for (String source : List.of(moduleSql(), unified)) { + int triggerStart = source.indexOf("CREATE TRIGGER trg_carrier_schedule_query_before_insert"); + int start = source.indexOf("WHERE booking.id=NEW.ocean_booking_id", triggerStart); + int end = source.indexOf(") THEN", start); + assertThat(triggerStart).isGreaterThanOrEqualTo(0); + assertThat(start).isGreaterThan(triggerStart); + assertThat(end).isGreaterThan(start); + String predicate = source.substring(start, end) + .replace("NEW.ocean_booking_id", "?").replace("NEW.customer_id", "?"); + String query = "SELECT EXISTS(SELECT 1 FROM fl_ocean_booking booking " + predicate + ")"; + assertThat(jdbc.queryForObject(query, Boolean.class, 1L, 0L)).isTrue(); + assertThat(jdbc.queryForObject(query, Boolean.class, 1L, 10001L)).isFalse(); + assertThat(jdbc.queryForObject(query, Boolean.class, 2L, 10001L)).isTrue(); + assertThat(jdbc.queryForObject(query, Boolean.class, 2L, 0L)).isFalse(); + assertThat(jdbc.queryForObject(query, Boolean.class, 2L, 10002L)).isFalse(); + assertThat(jdbc.queryForObject(query, Boolean.class, 99L, 0L)).isFalse(); + } + } + @Test void batchesActualContainerFactsWithoutEstimatedOrPerNodeQueries() throws Exception { String sql = String.join("\n", CarrierScheduleQueryMapper.class @@ -95,7 +125,7 @@ class CarrierScheduleQuerySqlContractTest { @Test void enforcesRealBookingContainerAndPermissions() throws Exception { assertThat(moduleSql()) - .contains("booking.id=NEW.ocean_booking_id AND booking.customer_id=NEW.customer_id") + .contains("booking.id=NEW.ocean_booking_id AND COALESCE(booking.customer_id,0)=NEW.customer_id") .contains("allocation.container_booking_id=NEW.container_booking_id") .contains("id IN (5102184,5102186,5102188)") .contains("platform:firstleg:port-schedule-query:manual")