package com.baiyue.lastleg.delivery; import com.baomidou.mybatisplus.core.MybatisConfiguration; import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean; import com.baiyue.lastleg.delivery.mapper.DeliveryFactsChangedMapper; import com.baiyue.lastleg.delivery.mapper.DeliveryOrderMapper; import com.baiyue.lastleg.delivery.repository.DeliveryFactsChangedRepository; import com.baiyue.lastleg.delivery.service.DeliveryFactsChangedService; import com.baiyue.lastleg.dispatch.port.DispatchDeliveryModels.Signal; import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mybatis.spring.SqlSessionTemplate; import org.springframework.aop.framework.ProxyFactory; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.datasource.DataSourceTransactionManager; import org.springframework.jdbc.datasource.DriverManagerDataSource; import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource; import org.springframework.transaction.interceptor.TransactionInterceptor; import org.springframework.transaction.support.TransactionTemplate; import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneId; import java.util.UUID; import static org.junit.jupiter.api.Assertions.*; /** 执行真实生产Mapper与事务,验证信号不能早确认或在回滚后遗留。 */ class DeliveryFactsChangedPersistenceTest { private JdbcTemplate db; private TransactionTemplate tx; private DeliveryFactsChangedService service; private final Instant now=Instant.parse("2026-10-19T01:00:00Z"); /** 每例建立隔离数据库,SQL写入仅存在于测试fixture。 */ @BeforeEach void prepare() throws Exception { var ds=new DriverManagerDataSource("jdbc:h2:mem:signals-"+UUID.randomUUID()+";MODE=MySQL;DB_CLOSE_DELAY=-1","sa",""); db=new JdbcTemplate(ds); db.execute(""" CREATE TABLE ll_delivery_facts_changed(id BIGINT PRIMARY KEY,customer_id VARCHAR(30),dispatch_id BIGINT, dispatch_no VARCHAR(50),delivery_id BIGINT,delivery_no VARCHAR(50),delivery_version BIGINT, facts_hash VARCHAR(64),occurred_at TIMESTAMP,state VARCHAR(20),attempt_count INT,next_attempt_at TIMESTAMP, last_error_code VARCHAR(64),processed_at TIMESTAMP,UNIQUE(customer_id,delivery_id,delivery_version)) """); db.execute(""" CREATE TABLE ll_delivery_order(id BIGINT PRIMARY KEY,customer_id VARCHAR(30),dispatch_order_id BIGINT, dispatch_no VARCHAR(50),delivery_no VARCHAR(50),version BIGINT,signed_at TIMESTAMP,signed_packages INT) """); db.update("INSERT INTO ll_delivery_order VALUES(1,'101',10,'DP-A','DL-A',1,?,5)",local(now)); var config=new MybatisConfiguration(); config.setMapUnderscoreToCamelCase(true); config.addMapper(DeliveryFactsChangedMapper.class); config.addMapper(DeliveryOrderMapper.class); var factory=new MybatisSqlSessionFactoryBean(); factory.setDataSource(ds); factory.setConfiguration(config); var session=new SqlSessionTemplate(factory.getObject()); var repository=new DeliveryFactsChangedRepository(session.getMapper(DeliveryFactsChangedMapper.class),session.getMapper(DeliveryOrderMapper.class)); var manager=new DataSourceTransactionManager(ds); tx=new TransactionTemplate(manager); tx.setIsolationLevel(org.springframework.transaction.TransactionDefinition.ISOLATION_READ_COMMITTED); var proxy=new ProxyFactory(new DeliveryFactsChangedService(repository,new ObjectMapper().findAndRegisterModules())); proxy.setProxyTargetClass(true); proxy.addAdvice(new TransactionInterceptor(manager,new AnnotationTransactionAttributeSource())); service=(DeliveryFactsChangedService)proxy.getProxy(); } /** 原签收事务失败则不存在候选,原请求重放只有一条事实。 */ @Test void enqueueRollsBackAndDeduplicatesWithRealMapper() { assertThrows(IllegalStateException.class,()->tx.execute(s->{service.enqueue("DP-A","DL-A",1,now);throw new IllegalStateException("rollback");})); assertTrue(service.due(now,20).isEmpty()); tx.executeWithoutResult(s->{service.enqueue("DP-A","DL-A",1,now);service.enqueue("DP-A","DL-A",1,now);}); assertEquals(1,service.due(now,20).size()); assertThrows(IllegalStateException.class,()->tx.executeWithoutResult(s->service.enqueue("DP-A","DL-A",2,now))); db.update("UPDATE ll_delivery_order SET signed_packages=4 WHERE id=1"); assertThrows(IllegalStateException.class,()->tx.executeWithoutResult(s->service.enqueue("DP-A","DL-A",1,now))); } /** 阶段写入与信号ack任一失败都回滚,失败记录保留可重试状态。 */ @Test void acknowledgmentRollsBackAndRetryNeverReopensProcessedFact() { tx.executeWithoutResult(s->service.enqueue("DP-A","DL-A",1,now)); Signal signal=service.due(now,20).get(0); assertThrows(IllegalStateException.class,()->tx.execute(s->{service.acknowledge(signal,now);throw new IllegalStateException("CAS failed");})); assertEquals(1,service.due(now,20).size()); service.failed(signal,now.plusSeconds(30),"DISPATCH_RECONCILE_FAILED"); assertTrue(service.due(now,20).isEmpty()); assertEquals(1,service.due(now.plusSeconds(30),20).get(0).attempts()); tx.executeWithoutResult(s->service.acknowledge(signal,now.plusSeconds(30))); service.failed(signal,now.plusSeconds(60),"DISPATCH_RECONCILE_FAILED"); assertTrue(service.due(now.plusSeconds(60),20).isEmpty()); assertEquals("PROCESSED",db.queryForObject("SELECT state FROM ll_delivery_facts_changed",String.class)); } /** 内部ID相同而业务身份不同也不能确认或记录别人的信号。 */ @Test void mismatchedIdentityAndMissingTransactionAreRejected() { tx.executeWithoutResult(s->service.enqueue("DP-A","DL-A",1,now)); var original=service.due(now,20).get(0); var other=new Signal(original.signalId(),"DP-B","DL-A",1,0,now); assertThrows(IllegalStateException.class,()->tx.executeWithoutResult(s->service.acknowledge(other,now))); assertThrows(org.springframework.transaction.IllegalTransactionStateException.class,()->service.acknowledge(original,now)); assertThrows(org.springframework.transaction.IllegalTransactionStateException.class,()->service.enqueue("DP-A","DL-A",1,now)); assertEquals(1,service.due(now,20).size()); } /** 与生产现有LocalDateTime字段保持服务端时区语义。 */ private static LocalDateTime local(Instant value) { return LocalDateTime.ofInstant(value,ZoneId.systemDefault()); } }