feat(fleet): enforce reviewed trips and original fee coverage for dispatch

2026-10-03 04:18:52 +08:00
parent c50e9f97d1
commit 5e683da3b9
162 changed files with 9642 additions and 218 deletions
@@ -58,6 +58,7 @@ public class OverseasFeeSource {
    private String reviewStatus;
    private String publicationStatus;
    private String sourceStatus;
+    @TableField(exist = false) private String voidStatus;
    private String reviewReason;
    private String reviewerId;
    private LocalDateTime reviewedAt;
@@ -15,6 +15,9 @@ import java.util.List;
@RequiredArgsConstructor
 public class OverseasFeeOutboxRepository {
    private final LastlegEventOutboxMapper mapper;
+    private final com.baiyue.lastleg.financefee.state.OverseasFeeCancellation cancellation;
+
+    public boolean prerequisitePublished(LastlegEventOutbox row) { return cancellation.prerequisitePublished(row); }

    public List<LastlegEventOutbox> pending(int limit, LocalDateTime now, LocalDateTime stale) {
        return mapper.selectList(new LambdaQueryWrapper<LastlegEventOutbox>()
@@ -22,6 +25,12 @@ public class OverseasFeeOutboxRepository {
            .and(status -> status.eq(LastlegEventOutbox::getStatus, "PENDING")
                .and(retry -> retry.isNull(LastlegEventOutbox::getNextRetryAt).or().le(LastlegEventOutbox::getNextRetryAt, now))
                .or(processing -> processing.eq(LastlegEventOutbox::getStatus, "PROCESSING").le(LastlegEventOutbox::getLockedAt, stale)))
+            .and(eligible -> eligible.eq(LastlegEventOutbox::getStatus, "PROCESSING")
+                .or().isNull(LastlegEventOutbox::getEventKey).or().notLikeLeft(LastlegEventOutbox::getEventKey, ":VOIDED")
+                .or().apply("EXISTS (SELECT 1 FROM ll_event_outbox prerequisite WHERE prerequisite.event_key IN ("
+                    + "CONCAT(SUBSTRING(ll_event_outbox.event_key,1,LENGTH(ll_event_outbox.event_key)-6),'PUBLISHED'),"
+                    + "CONCAT(SUBSTRING(ll_event_outbox.event_key,1,LENGTH(ll_event_outbox.event_key)-6),'ADJUSTED')) "
+                    + "AND prerequisite.status='PUBLISHED')"))
            .orderByAsc(LastlegEventOutbox::getCreatedAt).last("LIMIT " + limit));
    }
    public boolean claim(long id, String worker, LocalDateTime now) {
@@ -47,9 +56,12 @@ public class OverseasFeeOutboxRepository {
        return mapper.selectOne(new LambdaQueryWrapper<LastlegEventOutbox>().eq(LastlegEventOutbox::getId, id)
            .eq(LastlegEventOutbox::getStatus, "PROCESSING").eq(LastlegEventOutbox::getLockedBy, worker));
    }
+    @org.springframework.transaction.annotation.Transactional
    public int published(long id, String worker, LocalDateTime now) {
-        return mapper.update(null, claimed(id, worker, "PUBLISHED").set(LastlegEventOutbox::getPublishedAt, now)
+        int changed = mapper.update(null, claimed(id, worker, "PUBLISHED").set(LastlegEventOutbox::getPublishedAt, now)
            .set(LastlegEventOutbox::getNextRetryAt, null).set(LastlegEventOutbox::getLastError, null));
+        if (changed == 1) cancellation.acknowledge(mapper.selectById(id));
+        return changed;
    }
    public int dead(long id, String worker, int failures, String error) {
        return mapper.update(null, claimed(id, worker, "DEAD")
@@ -54,6 +54,7 @@ public class OverseasFeeOutboxRelay implements EventDeliveryProducer {
                        failures, next, "投递租约超时，消费结果未知"));
                continue; // Recovery always waits for the new backoff window; never sends in this pass.
            }
+            if (!repository.prerequisitePublished(candidate)) continue;
            String worker = UUID.randomUUID().toString();
            String type = Stream.FEE.type(candidate.getEventType());
            if (!journal.begin(worker, Stream.FEE.service, type, String.valueOf(candidate.getId()),
@@ -11,6 +11,7 @@ import com.baiyue.lastleg.financefee.entity.OverseasFeeCommand;
 import com.baiyue.lastleg.financefee.entity.OverseasFeeSource;
 import com.baiyue.lastleg.financefee.mapper.OverseasFeeCommandMapper;
 import com.baiyue.lastleg.financefee.mapper.OverseasFeeSourceMapper;
+import com.baiyue.lastleg.financefee.state.OverseasFeeCancellation;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.RequiredArgsConstructor;
@@ -41,31 +42,43 @@ public class OverseasFeeSourceService {
    private final OverseasFeeOutboxService outbox;
    private final OverseasFeeCurrencyConverter currencyConverter;
    private final ObjectMapper objectMapper;
+    private final OverseasFeeCancellation cancellation;

    public List<OverseasFeeSource> list(String customerId, String businessType, String businessNo,
                                        Long containerBookingId) {
-        return mapper.selectList(new LambdaQueryWrapper<OverseasFeeSource>()
+        return cancellation.describe(mapper.selectList(new LambdaQueryWrapper<OverseasFeeSource>()
            .eq(text(customerId) != null, OverseasFeeSource::getCustomerId, text(customerId))
            .eq(text(businessType) != null, OverseasFeeSource::getBusinessType, text(businessType))
            .like(text(businessNo) != null, OverseasFeeSource::getBusinessNo, text(businessNo))
            .eq(containerBookingId != null, OverseasFeeSource::getContainerBookingId, containerBookingId)
            .inSql(OverseasFeeSource::getId, "SELECT MAX(v.id) FROM ll_finance_fee_source v GROUP BY v.source_fee_id")
-            .orderByDesc(OverseasFeeSource::getCreatedAt).orderByDesc(OverseasFeeSource::getId));
+            .orderByDesc(OverseasFeeSource::getCreatedAt).orderByDesc(OverseasFeeSource::getId)));
    }

    public List<OverseasFeeSource> listForBusiness(String customerId, String businessType, long businessId) {
-        return mapper.selectList(new LambdaQueryWrapper<OverseasFeeSource>()
+        return cancellation.describe(mapper.selectList(new LambdaQueryWrapper<OverseasFeeSource>()
            .eq(OverseasFeeSource::getCustomerId, customerId)
            .eq(OverseasFeeSource::getBusinessType, businessType)
            .eq(OverseasFeeSource::getBusinessId, businessId)
            .inSql(OverseasFeeSource::getId, "SELECT MAX(v.id) FROM ll_finance_fee_source v GROUP BY v.source_fee_id")
-            .orderByAsc(OverseasFeeSource::getCreatedAt).orderByAsc(OverseasFeeSource::getId));
+            .orderByAsc(OverseasFeeSource::getCreatedAt).orderByAsc(OverseasFeeSource::getId)));
    }

    public List<ContainerCustomerOption> listContainerCustomers(long containerBookingId) {
        return businessGuard.listContainerCustomers(containerBookingId);
    }

+    public void requireNoActiveDeliveryReceivables(String customerId, List<String> inboundCodes) {
+        if (inboundCodes.isEmpty()) return;
+        long existing = mapper.selectCount(new LambdaQueryWrapper<OverseasFeeSource>()
+            .eq(OverseasFeeSource::getCustomerId, customerId).eq(OverseasFeeSource::getBusinessType, "DELIVERY_ORDER")
+            .eq(OverseasFeeSource::getDirection, "RECEIVABLE").eq(OverseasFeeSource::getFeeItemCode, "DELIVERY_FEE")
+            .in(OverseasFeeSource::getInboundCode, inboundCodes).ne(OverseasFeeSource::getSourceStatus, "VOIDED")
+            .and(state -> state.eq(OverseasFeeSource::getReviewStatus, "APPROVED").or().gt(OverseasFeeSource::getSourceVersion, 1))
+            .inSql(OverseasFeeSource::getId, "SELECT MAX(v.id) FROM ll_finance_fee_source v GROUP BY v.source_fee_id"));
+        if (existing > 0) throw conflict("外接货已有派送应收尚未确认作废，请先完成原费用处理再重新转派送");
+    }
+
    @Transactional
    public OverseasFeeSource create(String operatorId, String key, SaveRequest request) {
        OverseasFeeCommand replay = replay(operatorId, "CREATE", key, request);
@@ -125,11 +138,12 @@ public class OverseasFeeSourceService {
        return value;
    }

-    @Transactional
+    @Transactional(isolation = org.springframework.transaction.annotation.Isolation.READ_COMMITTED)
    public OverseasFeeSource adjust(long sourceFeeId, String operatorId, AdjustRequest request) {
        OverseasFeeCommand replay = replay(operatorId, "ADJUST", request.idempotencyKey(), request);
        if (replay != null) return current(replay.getSourceFeeId());
        OverseasFeeSource old = lock(sourceFeeId);
+        if (cancellation.requested(old)) throw conflict("该费用已申请作废，请先处理财务结果");
        if (!"APPROVED".equals(old.getReviewStatus()) || !"ACTIVE".equals(old.getSourceStatus())
            || old.getSourceVersion() != request.expectedVersion()) throw conflict("只有当前已发布版本可以调整");
        validate(request.fee()); businessGuard.requireOwned(request.fee());
@@ -157,10 +171,15 @@ public class OverseasFeeSourceService {
        OverseasFeeSource value = lock(sourceFeeId);
        if (!"APPROVED".equals(value.getReviewStatus()) || !"ACTIVE".equals(value.getSourceStatus())
            || value.getSourceVersion() != request.expectedVersion()) throw conflict("只有当前已发布版本可以作废");
-        value.setSourceStatus("VOIDED"); value.setReviewReason(request.reason().trim());
+        if (cancellation.requested(value)) return cancellation.describe(List.of(value)).get(0);
+        value.setReviewReason(request.reason().trim());
        value.setUpdatedBy(operatorId); value.setUpdatedAt(now()); mapper.updateById(value);
-        outbox.publish(value, "VOIDED");
+        var pending = new OverseasFeeSource();
+        org.springframework.beans.BeanUtils.copyProperties(value, pending);
+        pending.setSourceStatus("VOIDED");
+        outbox.publish(pending, "VOIDED");
        command(operatorId, "VOID", request.idempotencyKey(), request, sourceFeeId, value.getSourceVersion());
+        value.setVoidStatus("PENDING_FINANCE");
        return value;
    }

@@ -233,7 +252,7 @@ public class OverseasFeeSourceService {
    private OverseasFeeSource current(long sourceFeeId) {
        OverseasFeeSource value = mapper.selectOne(new LambdaQueryWrapper<OverseasFeeSource>()
            .eq(OverseasFeeSource::getSourceFeeId, sourceFeeId).orderByDesc(OverseasFeeSource::getSourceVersion).last("LIMIT 1"));
-        if (value == null) throw notFound(); return value;
+        if (value == null) throw notFound(); return cancellation.describe(List.of(value)).get(0);
    }
    private OverseasFeeSource lock(long sourceFeeId) { OverseasFeeSource value = mapper.lockCurrent(sourceFeeId); if (value == null) throw notFound(); return value; }
    private OverseasFeeCommand replay(String operator, String action, String key, Object request) {
@@ -0,0 +1,80 @@
+package com.baiyue.lastleg.financefee.state;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baiyue.common.event.BusinessFeeSourceChangedV1;
+import com.baiyue.common.event.EventEnvelope;
+import com.baiyue.common.event.EventTypes;
+import com.baiyue.lastleg.entity.LastlegEventOutbox;
+import com.baiyue.lastleg.financefee.entity.OverseasFeeSource;
+import com.baiyue.lastleg.financefee.mapper.OverseasFeeSourceMapper;
+import com.baiyue.lastleg.mapper.LastlegEventOutboxMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/** A request is not a financial void. Only the matched durable consumer ACK confirms it. */
+@Component
+@RequiredArgsConstructor
+public class OverseasFeeCancellation {
+    private final LastlegEventOutboxMapper events;
+    private final OverseasFeeSourceMapper fees;
+    private final ObjectMapper json;
+
+    public static String eventKey(OverseasFeeSource fee, String action) {
+        return EventTypes.BUSINESS_FEE_SOURCE_CHANGED + ":LASTLEG:" + fee.getSourceType() + ":"
+            + fee.getSourceFeeId() + ":" + fee.getSourceVersion() + ":" + action;
+    }
+    public boolean requested(OverseasFeeSource fee) { return events.findByEventKey(eventKey(fee, "VOIDED")) != null; }
+
+    public List<OverseasFeeSource> describe(List<OverseasFeeSource> values) {
+        if (values.isEmpty()) return values;
+        var keys = values.stream().flatMap(value -> java.util.stream.Stream.of(eventKey(value, "VOIDED"),
+            eventKey(value, value.getSourceVersion() == 1 ? "PUBLISHED" : "ADJUSTED"))).toList();
+        var byKey = events.selectList(new LambdaQueryWrapper<LastlegEventOutbox>().in(LastlegEventOutbox::getEventKey, keys))
+            .stream().collect(Collectors.toMap(LastlegEventOutbox::getEventKey, row -> row));
+        for (var value : values) {
+            var request = byKey.get(eventKey(value, "VOIDED"));
+            var publish = byKey.get(eventKey(value, value.getSourceVersion() == 1 ? "PUBLISHED" : "ADJUSTED"));
+            value.setVoidStatus("VOIDED".equals(value.getSourceStatus()) ? "CONFIRMED" : request == null ? "NOT_REQUESTED"
+                : "DEAD".equals(request.getStatus()) || publish == null || "DEAD".equals(publish.getStatus())
+                    ? "FINANCE_ATTENTION" : "PENDING_FINANCE");
+        }
+        return values;
+    }
+
+    public boolean prerequisitePublished(LastlegEventOutbox request) {
+        if (request.getEventKey() == null || !request.getEventKey().endsWith(":VOIDED")) return true;
+        String prefix = request.getEventKey().substring(0, request.getEventKey().length() - "VOIDED".length());
+        return events.selectCount(new LambdaQueryWrapper<LastlegEventOutbox>()
+            .in(LastlegEventOutbox::getEventKey, List.of(prefix + "PUBLISHED", prefix + "ADJUSTED"))
+            .eq(LastlegEventOutbox::getStatus, "PUBLISHED")) > 0;
+    }
+
+    /** Invoked inside the same transaction as the lease-owner CAS and delivery journal completion. */
+    @Transactional(propagation = Propagation.MANDATORY)
+    public void acknowledge(LastlegEventOutbox event) {
+        if (event.getEventKey() == null || !event.getEventKey().endsWith(":VOIDED")) return;
+        try {
+            var envelope = json.readValue(event.getPayloadJson(), EventEnvelope.class).requireValid();
+            var payload = json.treeToValue(envelope.payload(), BusinessFeeSourceChangedV1.class);
+            if (!"VOIDED".equals(payload.action()) || !"LASTLEG".equals(payload.sourceSystem()))
+                throw new IllegalStateException("费用作废事件与来源不一致");
+            var source = fees.lockCurrent(Long.parseLong(payload.sourceFeeId()));
+            if (source == null || !Objects.equals(source.getSourceVersion(), payload.sourceVersion())
+                || !Objects.equals(source.getCustomerId(), payload.customerId())
+                || !Objects.equals(source.getSourceType(), payload.sourceType())
+                || !eventKey(source, "VOIDED").equals(event.getEventKey()) || !"ACTIVE".equals(source.getSourceStatus()))
+                throw new IllegalStateException("财务已确认的作废来源与当前业务版本不一致");
+            source.setSourceStatus("VOIDED"); source.setUpdatedAt(LocalDateTime.now(ZoneOffset.UTC));
+            if (fees.updateById(source) != 1) throw new IllegalStateException("费用作废确认未能落账");
+        } catch (com.fasterxml.jackson.core.JsonProcessingException error) {
+            throw new IllegalStateException("费用作废事件无法解析", error);
+        }
+    }
+}