baiyue-firstleg/event-bridge-service/src/main/java/com/baiyue/eventbridge/service/EventInboxService.java

package com.baiyue.eventbridge.service;

import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
import com.baiyue.eventbridge.domain.BridgeInbox;
import com.baiyue.eventbridge.dto.EventReceiveResult;
import com.baiyue.eventbridge.metrics.EventBridgeMetrics;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import jakarta.annotation.Resource;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.Objects;

/**
 * 入站事件校验与幂等落库服务。
 */
@Service
public class EventInboxService {
    @Resource private InboxStore inboxStore;
    @Resource private ObjectMapper objectMapper;
    @Resource private EventBridgeMetrics metrics;

    @Value("${baiyue.region:CN}")
    private String localRegion;

    @Transactional
    public EventReceiveResult receive(EventEnvelope envelope) {
        envelope.requireValid();
        if (!envelope.targetRegion().name().equalsIgnoreCase(localRegion)) {
            throw new IllegalArgumentException("事件目标区域与当前事件桥区域不一致");
        }
        if (!EventPayloadHash.matches(
                EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) {
            throw new IllegalArgumentException("payloadHash 与规范化事件负载不一致");
        }
        BridgeInbox existing = inboxStore.findByEventId(envelope.eventId());
        if (existing != null) {
            return duplicateOrConflict(existing, envelope);
        }
        String idempotencyKey = idempotencyKey(envelope);
        BridgeInbox sameBusinessEvent = inboxStore.findByIdempotencyKey(idempotencyKey);
        if (sameBusinessEvent != null) {
            return duplicateBusinessEventOrConflict(sameBusinessEvent, envelope);
        }
        BridgeInbox inbox = toInbox(envelope);
        try {
            inboxStore.insert(inbox);
            metrics.recordReceive(envelope, EventBridgeMetrics.ReceiveResult.RECEIVED);
            return new EventReceiveResult(envelope.eventId(), "已接收", inbox.getReceivedAt());
        } catch (DuplicateKeyException exception) {
            BridgeInbox concurrent = inboxStore.findByEventId(envelope.eventId());
            if (concurrent == null) concurrent = inboxStore.findByIdempotencyKey(idempotencyKey);
            if (concurrent == null) {
                throw exception;
            }
            return Objects.equals(concurrent.getEventId(), envelope.eventId())
                ? duplicateOrConflict(concurrent, envelope)
                : duplicateBusinessEventOrConflict(concurrent, envelope);
        }
    }

    private EventReceiveResult duplicateBusinessEventOrConflict(BridgeInbox existing, EventEnvelope envelope) {
        if (!sameBusinessEvent(existing, envelope)) {
            metrics.recordReceive(envelope, EventBridgeMetrics.ReceiveResult.CONFLICT);
            throw new IllegalArgumentException("相同eventKey对应的事件负载不一致，拒绝柜号或事实漂移");
        }
        metrics.recordReceive(envelope, EventBridgeMetrics.ReceiveResult.DUPLICATE);
        return new EventReceiveResult(existing.getEventId(), "重复", existing.getReceivedAt());
    }

    private EventReceiveResult duplicateOrConflict(BridgeInbox existing, EventEnvelope envelope) {
        if (!sameEnvelope(existing, envelope)) {
            metrics.recordReceive(envelope, EventBridgeMetrics.ReceiveResult.CONFLICT);
            throw new IllegalArgumentException("相同 eventId 对应的事件信封或 payloadHash 不一致，拒绝覆盖既有事件");
        }
        metrics.recordReceive(envelope, EventBridgeMetrics.ReceiveResult.DUPLICATE);
        return new EventReceiveResult(envelope.eventId(), "重复", existing.getReceivedAt());
    }

    private boolean sameEnvelope(BridgeInbox existing, EventEnvelope envelope) {
        return sameBusinessEvent(existing, envelope)
            && Objects.equals(normalizeOccurredAt(existing.getOccurredAt()), normalizeOccurredAt(envelope.occurredAt()))
            && Objects.equals(existing.getTraceId(), envelope.traceId())
            && Objects.equals(existing.getCausationId(), envelope.causationId());
    }

    private boolean sameBusinessEvent(BridgeInbox existing, EventEnvelope envelope) {
        return existing.getPayloadHash() != null
            && existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())
            && Objects.equals(existing.getEventType(), envelope.eventType())
            && Objects.equals(existing.getEventVersion(), envelope.eventVersion())
            && Objects.equals(existing.getSourceRegion(), envelope.sourceRegion().name())
            && Objects.equals(existing.getTargetRegion(), envelope.targetRegion().name())
            && Objects.equals(existing.getCustomerId(), envelope.tenantId())
            && Objects.equals(existing.getAggregateType(), envelope.aggregateType())
            && Objects.equals(existing.getAggregateId(), envelope.aggregateId())
            && Objects.equals(existing.getAggregateVersion(), envelope.aggregateVersion())
            && Objects.equals(existing.getHandoverVersion(), envelope.handoverVersion())
            && Objects.equals(existing.getCorrelationId(), envelope.correlationId());
    }

    private BridgeInbox toInbox(EventEnvelope envelope) {
        BridgeInbox inbox = new BridgeInbox();
        inbox.setEventId(envelope.eventId());
        inbox.setIdempotencyKey(idempotencyKey(envelope));
        inbox.setEventType(envelope.eventType());
        inbox.setEventVersion(envelope.eventVersion());
        inbox.setSourceRegion(envelope.sourceRegion().name());
        inbox.setTargetRegion(envelope.targetRegion().name());
        inbox.setCustomerId(envelope.tenantId());
        inbox.setAggregateType(envelope.aggregateType());
        inbox.setAggregateId(envelope.aggregateId());
        inbox.setAggregateVersion(envelope.aggregateVersion());
        inbox.setHandoverVersion(envelope.handoverVersion());
        inbox.setTraceId(envelope.traceId());
        inbox.setCorrelationId(envelope.correlationId());
        inbox.setCausationId(envelope.causationId());
        inbox.setOccurredAt(normalizeOccurredAt(envelope.occurredAt()));
        inbox.setPayloadHash(envelope.payloadHash().toLowerCase(java.util.Locale.ROOT));
        try {
            inbox.setPayloadJson(objectMapper.writeValueAsString(envelope.payload()));
        } catch (JsonProcessingException exception) {
            throw new IllegalArgumentException("事件负载无法序列化", exception);
        }
        inbox.setStatus("RECEIVED");
        inbox.setReceivedAt(LocalDateTime.now());
        return inbox;
    }

    private static Instant normalizeOccurredAt(Instant value) {
        return value == null ? null : value.truncatedTo(ChronoUnit.MILLIS);
    }

    private static String idempotencyKey(EventEnvelope envelope) {
        return envelope.sourceRegion() + ":" + envelope.correlationId();
    }
}