fix(firstleg): serialize concurrent customs start replay

firstleg/src/main/java/com/baiyue/firstleg/customs/container/service/ContainerCustomsService.java

@@ -149,12 +149,12 @@ public class ContainerCustomsService {
 
     @Transactional
     public ContainerView start(long planId, long operatorId, String operatorName, String key, ActionRequest request) {
-        ContainerRow current = ensure(planId);
         String action = "START_CONTAINER_CUSTOMS_" + planId;
         if(!request.confirm())throw bad("请确认开始整柜报关");
         String requestHash = hash(action, request.expectedVersion(), request.confirm(), request.remark());
         CommandRow replay = claim(operatorId, action, key, requestHash);
         if (replay != null) return platformView(planId);
+        ensure(planId);
         ContainerRow locked = mapper.lockContainer(planId);
         if (locked == null || !READY.equals(locked.status())) throw conflict("当前柜报关资料尚未齐全或已开始报关");
         if (!Objects.equals(locked.version(), request.expectedVersion())) throw concurrent();

firstleg/src/test/java/com/baiyue/firstleg/customs/container/ContainerCustomsServiceTest.java

@@ -13,6 +13,10 @@ import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.mock.web.MockMultipartFile;
 
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -104,6 +108,70 @@ class ContainerCustomsServiceTest {
         verify(mapper,never()).freezeScopes(anyLong());
     }
 
+    @Test void concurrentStartWithSameKeyReplaysOneAggregateAndDifferentPayloadConflicts() throws Exception {
+        var row=new ContainerCustomsMapper.InboundRow(9L,"LP9",8L,"COMPLETED","CAB","C",11L,21L,"O","I",7L,"客户","TAX_REFUND");
+        var ready=container(ContainerCustomsService.READY,2);
+        var declaring=container(ContainerCustomsService.DECLARING,3);
+        var aggregateState=new AtomicReference<>(ready);
+        var commandState=new AtomicReference<ContainerCustomsMapper.CommandRow>();
+        var claimCalls=new AtomicInteger();
+        var secondClaimEntered=new CountDownLatch(1);
+        var commandCompleted=new CountDownLatch(1);
+        when(mapper.planInbounds(9L)).thenReturn(List.of(row));
+        when(mapper.container(9L)).thenAnswer(invocation->aggregateState.get());
+        when(mapper.missingRefundCount(9L,2L)).thenReturn(0);
+        when(mapper.claimCommand(anyLong(),eq(5L),eq("START_CONTAINER_CUSTOMS_9"),eq("start-key"),anyString()))
+            .thenAnswer(invocation->{
+                int call=claimCalls.incrementAndGet();
+                if(call==1){
+                    commandState.set(new ContainerCustomsMapper.CommandRow(invocation.getArgument(0),5L,
+                        "START_CONTAINER_CUSTOMS_9","start-key",invocation.getArgument(4),"PROCESSING",null,null,null));
+                    assertThat(secondClaimEntered.await(5,TimeUnit.SECONDS)).isTrue();
+                    return 1;
+                }
+                secondClaimEntered.countDown();
+                assertThat(commandCompleted.await(5,TimeUnit.SECONDS)).isTrue();
+                return 0;
+            });
+        when(mapper.command(5L,"START_CONTAINER_CUSTOMS_9","start-key")).thenAnswer(invocation->commandState.get());
+        when(mapper.lockContainer(9L)).thenAnswer(invocation->aggregateState.get());
+        when(mapper.start(2L,2,5L,"admin")).thenAnswer(invocation->{aggregateState.set(declaring);return 1;});
+        when(mapper.completeCommand(eq(5L),eq("START_CONTAINER_CUSTOMS_9"),eq("start-key"),eq(2L)))
+            .thenAnswer(invocation->{
+                var current=commandState.get();
+                commandState.set(new ContainerCustomsMapper.CommandRow(current.id(),current.actorId(),current.action(),
+                    current.idempotencyKey(),current.requestHash(),"SUCCESS",2L,null,null));
+                commandCompleted.countDown();
+                return 1;
+            });
+        when(mapper.documents(2L,null)).thenReturn(List.of());
+
+        var request=new ActionRequest(2,true,"same");
+        var executor=Executors.newFixedThreadPool(2);
+        try{
+            var first=executor.submit(()->service.start(9L,5L,"admin","start-key",request));
+            var second=executor.submit(()->service.start(9L,5L,"admin","start-key",request));
+            var firstResult=first.get(8,TimeUnit.SECONDS);
+            var secondResult=second.get(8,TimeUnit.SECONDS);
+
+            assertThat(firstResult.aggregateId()).isEqualTo(2L);
+            assertThat(secondResult.aggregateId()).isEqualTo(2L);
+            assertThat(firstResult.status()).isEqualTo(ContainerCustomsService.DECLARING);
+            assertThat(secondResult.status()).isEqualTo(ContainerCustomsService.DECLARING);
+        }finally{
+            executor.shutdownNow();
+        }
+
+        assertThatThrownBy(()->service.start(9L,5L,"admin","start-key",new ActionRequest(2,true,"different")))
+            .isInstanceOfSatisfying(BizException.class,
+                error->assertThat(error.getResultCode()).isEqualTo(ResultCode.CONFLICT));
+        InOrder order=inOrder(mapper);
+        order.verify(mapper,times(2)).claimCommand(anyLong(),eq(5L),eq("START_CONTAINER_CUSTOMS_9"),eq("start-key"),anyString());
+        order.verify(mapper,atLeastOnce()).planInbounds(9L);
+        verify(mapper,times(1)).start(2L,2,5L,"admin");
+        verify(mapper,times(1)).completeCommand(5L,"START_CONTAINER_CUSTOMS_9","start-key",2L);
+    }
+
     @Test void lastCompletionRequiresConfirmationBeforeCallingFile(){
         var source=document(1L);var container=container(ContainerCustomsService.DECLARING,3);
         when(mapper.claimCommand(anyLong(),eq(5L),eq("UPLOAD_COMPLETION_1"),eq("k"),anyString())).thenReturn(1);