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.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 java.time.LocalDateTime; /** * 入站事件校验与幂等落库服务。 */ @Service public class EventInboxService { private final InboxStore inboxStore; private final ObjectMapper objectMapper; private final String localRegion; public EventInboxService(InboxStore inboxStore, ObjectMapper objectMapper, @Value("${baiyue.region:CN}") String localRegion) { this.inboxStore = inboxStore; this.objectMapper = objectMapper; this.localRegion = 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); } BridgeInbox inbox = toInbox(envelope); try { inboxStore.insert(inbox); return new EventReceiveResult(envelope.eventId(), "已接收", inbox.getReceivedAt()); } catch (DuplicateKeyException exception) { BridgeInbox concurrent = inboxStore.findByEventId(envelope.eventId()); if (concurrent == null) { throw exception; } return duplicateOrConflict(concurrent, envelope); } } private EventReceiveResult duplicateOrConflict(BridgeInbox existing, EventEnvelope envelope) { if (!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) { throw new IllegalArgumentException("相同 eventId 对应的 payloadHash 不一致,拒绝覆盖既有事件"); } return new EventReceiveResult(envelope.eventId(), "重复", existing.getReceivedAt()); } private BridgeInbox toInbox(EventEnvelope envelope) { BridgeInbox inbox = new BridgeInbox(); inbox.setEventId(envelope.eventId()); inbox.setIdempotencyKey(envelope.sourceRegion() + ":" + envelope.eventId()); inbox.setEventType(envelope.eventType()); inbox.setEventVersion(envelope.eventVersion()); inbox.setSourceRegion(envelope.sourceRegion().name()); inbox.setTargetRegion(envelope.targetRegion().name()); inbox.setTenantId(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(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; } }