OutboxPublisherIntegrationTest.java

package com.beachcheck.outbox.integration;

import static com.beachcheck.notification.domain.Notification.NotificationStatus;
import static com.beachcheck.notification.domain.Notification.NotificationType;
import static com.beachcheck.outbox.domain.OutboxEvent.OutboxEventStatus;
import static com.beachcheck.outbox.domain.OutboxEvent.OutboxEventType;
import static com.beachcheck.support.tracing.SpanTestSupport.awaitSpans;
import static com.beachcheck.support.tracing.SpanTestSupport.flushSpans;
import static java.time.Instant.now;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;

import com.beachcheck.notification.domain.Notification;
import com.beachcheck.notification.repository.NotificationRepository;
import com.beachcheck.outbox.domain.OutboxEvent;
import com.beachcheck.outbox.repository.OutboxEventRepository;
import com.beachcheck.outbox.service.OutboxEventDispatcher;
import com.beachcheck.outbox.service.OutboxPublisher;
import com.beachcheck.outbox.tracing.OutboxTracing;
import com.beachcheck.support.base.IntegrationTest;
import com.beachcheck.support.tracing.RecordingSpanExporter;
import com.beachcheck.support.tracing.TracingTestConfiguration;
import com.beachcheck.user.domain.User;
import com.beachcheck.user.repository.UserRepository;
import com.google.firebase.messaging.FirebaseMessaging;
import com.google.firebase.messaging.FirebaseMessagingException;
import com.google.firebase.messaging.Message;
import com.google.firebase.messaging.MessagingErrorCode;
import io.micrometer.tracing.Tracer;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.trace.SpanKind;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.data.SpanData;
import java.time.Instant;
import java.util.List;
import java.util.UUID;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.test.context.TestPropertySource;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

/**
 * OutboxPublisher 통합 테스트
 *
 * <p>Why: 실제 DB, 실제 Bean을 사용하여 OutboxPublisher의 전체 플로우를 검증
 *
 * <p>Policy:
 *
 * <ul>
 *   <li>NOT_SUPPORTED: OutboxEventDispatcher의 REQUIRES_NEW와 격리. 롤백 대신 @BeforeEach deleteAll()로 테스트
 *       격리
 *   <li>FirebaseMessaging만 Mock (실제 FCM 호출 방지)
 * </ul>
 */
@Transactional(
    propagation =
        Propagation
            .NOT_SUPPORTED) // OutboxPublisher의 REQUIRES_NEW 트랜잭션과 격리하여 테스트 간 트랜잭션 롤백이 영향을 주지 않도록 함.
// 대신 @BeforeEach에서 deleteAll()로 DB 상태 초기화
@Import({TracingTestConfiguration.class, OutboxPublisherIntegrationTest.OutboxTestConfig.class})
@TestPropertySource(properties = "management.tracing.sampling.probability=1.0")
class OutboxPublisherIntegrationTest extends IntegrationTest {

  private static final String PRODUCER_TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736";
  private static final String PRODUCER_SPAN_ID = "00f067aa0ba902b7";
  private static final String PRODUCER_TRACEPARENT =
      "00-" + PRODUCER_TRACE_ID + "-" + PRODUCER_SPAN_ID + "-01";

  @Autowired private OutboxPublisher outboxPublisher;

  @Autowired private OutboxEventRepository outboxEventRepository;

  @Autowired private NotificationRepository notificationRepository;

  @Autowired private UserRepository userRepository;

  @Autowired private FirebaseMessaging firebaseMessaging; // 외부 서비스는 Mock 처리 (E2E 테스트에서는 실제 호출)

  @Autowired private RecordingSpanExporter exporter;

  @Autowired private SdkTracerProvider tracerProvider;

  @Autowired private Tracer tracer;

  @TestConfiguration(proxyBeanMethods = false)
  static class OutboxTestConfig {

    @Bean
    OutboxEventDispatcher outboxEventDispatcher(
        OutboxEventRepository outboxEventRepository,
        NotificationRepository notificationRepository,
        FirebaseMessaging firebaseMessaging,
        Tracer tracer) {
      return new OutboxEventDispatcher(
          outboxEventRepository, notificationRepository, firebaseMessaging, tracer);
    }

    @Bean
    OutboxPublisher outboxPublisher(
        OutboxEventRepository outboxEventRepository,
        OutboxEventDispatcher outboxEventDispatcher,
        OutboxTracing outboxTracing,
        @Value("${app.outbox.polling.batch-size:10}") int batchSize) {
      return new OutboxPublisher(
          outboxEventRepository, outboxEventDispatcher, outboxTracing, batchSize);
    }
  }

  @BeforeEach
  void setUp() throws FirebaseMessagingException {
    flushSpans(tracerProvider);
    exporter.clear();

    // Mock 호출 기록 초기화 (테스트 간 격리)
    clearInvocations(firebaseMessaging);

    // NOT_SUPPORTED로 외부 트랜잭션이 없어 롤백이 동작하지 않으므로 명시적 정리 필요
    // OutboxEventDispatcher의 REQUIRES_NEW 커밋 데이터도 롤백 대상 외라 deleteAll()이 실질적 격리 수단
    // FK 제약 조건 순서 고려하여 삭제
    outboxEventRepository.deleteAll();
    notificationRepository.deleteAll();
    userRepository.deleteAll();

    // FirebaseMessaging Mock 기본 동작 설정
    given(firebaseMessaging.send(any(Message.class))).willReturn("mock-message-id");
  }

  @Test
  @DisplayName("TC1 - PENDING 이벤트를 폴링하여 FCM 전송 후 SENT 상태로 전이")
  void shouldProcessPendingEventAndMarkAsSent() throws FirebaseMessagingException {
    // Given: 실제 DB에 Notification + OutboxEvent 저장
    Notification notification = createAndSaveNotification(NotificationStatus.PENDING);
    OutboxEvent event = createAndSaveOutboxEvent(notification.getId());
    resetSpans();

    // When: OutboxPublisher 수동 실행 (스케줄러 대신 직접 호출하여 테스트 제어)
    outboxPublisher.processPendingOutboxEvents();

    // Then: 이벤트 상태 확인
    OutboxEvent processedEvent =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(processedEvent.getStatus()).isEqualTo(OutboxEventStatus.SENT);
    assertThat(processedEvent.getProcessedAt()).isNotNull();

    // Then: Notification 상태 확인
    Notification processedNotification =
        notificationRepository
            .findById(notification.getId())
            .orElseThrow(() -> new IllegalStateException("Notification을 찾을 수 없습니다"));
    assertThat(processedNotification.getStatus()).isEqualTo(NotificationStatus.SENT);
    assertThat(processedNotification.getSentAt()).isNotNull();

    // Then: FCM 전송 확인
    then(firebaseMessaging).should().send(any(Message.class));

    // Then: runtime trace 구조와 성공 결과 확인
    List<SpanData> spans = awaitSpans(exporter, tracerProvider, 5);
    SpanData processSpan = onlySpanNamed(spans, "outbox process");
    SpanData dispatchSpan = onlySpanNamed(spans, "outbox dispatch");
    SpanData lookupSpan = onlySpanNamed(spans, "outbox notification lookup");
    SpanData fcmSpan = onlySpanNamed(spans, "outbox fcm send");
    SpanData stateSpan = onlySpanNamed(spans, "outbox state update");

    assertThat(dispatchSpan.getParentSpanId()).isEqualTo(processSpan.getSpanId());
    assertThat(lookupSpan.getParentSpanId()).isEqualTo(dispatchSpan.getSpanId());
    assertThat(fcmSpan.getParentSpanId()).isEqualTo(dispatchSpan.getSpanId());
    assertThat(stateSpan.getParentSpanId()).isEqualTo(dispatchSpan.getSpanId());
    assertThat(dispatchSpan.getKind()).isEqualTo(SpanKind.INTERNAL);
    assertThat(lookupSpan.getKind()).isEqualTo(SpanKind.INTERNAL);
    assertThat(fcmSpan.getKind()).isEqualTo(SpanKind.CLIENT);
    assertThat(stateSpan.getKind()).isEqualTo(SpanKind.INTERNAL);
    assertThat(dispatchSpan.getAttributes().get(AttributeKey.stringKey("outbox.item.outcome")))
        .isEqualTo("success");
    assertThat(fcmSpan.getAttributes().get(AttributeKey.stringKey("outbox.fcm.outcome")))
        .isEqualTo("success");
    assertThat(tracer.currentSpan()).isNull();
    assertNoSensitiveTraceData(spans, notification);
  }

  @Test
  @DisplayName("TC2 - 여러 PENDING 이벤트를 배치 처리")
  void shouldProcessMultiplePendingEvents() {
    // Given: 3개의 PENDING 이벤트 생성
    Notification notification1 = createAndSaveNotification(NotificationStatus.PENDING);
    Notification notification2 = createAndSaveNotification(NotificationStatus.PENDING);
    Notification notification3 = createAndSaveNotification(NotificationStatus.PENDING);

    OutboxEvent event1 = createAndSaveOutboxEvent(notification1.getId());
    OutboxEvent event2 = createAndSaveOutboxEvent(notification2.getId());
    OutboxEvent event3 = createAndSaveOutboxEvent(notification3.getId());

    // When: OutboxPublisher 실행
    outboxPublisher.processPendingOutboxEvents();

    // Then: 생성한 이벤트 각각 SENT 상태 확인 (findAll() 대신 ID 기반으로 전역 데이터 오염 방지)
    assertThat(outboxEventRepository.findById(event1.getId()))
        .get()
        .extracting(OutboxEvent::getStatus)
        .isEqualTo(OutboxEventStatus.SENT);
    assertThat(outboxEventRepository.findById(event2.getId()))
        .get()
        .extracting(OutboxEvent::getStatus)
        .isEqualTo(OutboxEventStatus.SENT);
    assertThat(outboxEventRepository.findById(event3.getId()))
        .get()
        .extracting(OutboxEvent::getStatus)
        .isEqualTo(OutboxEventStatus.SENT);

    // Then: 생성한 Notification 각각 SENT 상태 확인
    assertThat(notificationRepository.findById(notification1.getId()))
        .get()
        .extracting(Notification::getStatus)
        .isEqualTo(NotificationStatus.SENT);
    assertThat(notificationRepository.findById(notification2.getId()))
        .get()
        .extracting(Notification::getStatus)
        .isEqualTo(NotificationStatus.SENT);
    assertThat(notificationRepository.findById(notification3.getId()))
        .get()
        .extracting(Notification::getStatus)
        .isEqualTo(NotificationStatus.SENT);
  }

  @Test
  @DisplayName("TC3 - 멱등성: 이미 SENT 상태면 FCM 전송 스킵")
  void shouldSkipAlreadySentNotification() throws FirebaseMessagingException {
    // Given: SENT 상태의 Notification + PENDING OutboxEvent
    Notification notification = createAndSaveNotification(NotificationStatus.SENT);
    OutboxEvent event = createAndSaveOutboxEvent(notification.getId());
    resetSpans();

    // When: OutboxPublisher 실행
    outboxPublisher.processPendingOutboxEvents();

    // Then: OutboxEvent는 SENT로 전이
    OutboxEvent processedEvent =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(processedEvent.getStatus()).isEqualTo(OutboxEventStatus.SENT);

    // Then: FCM 전송 안 됨
    then(firebaseMessaging).should(never()).send(any(Message.class));

    List<SpanData> spans = awaitSpans(exporter, tracerProvider, 4);
    SpanData dispatchSpan = onlySpanNamed(spans, "outbox dispatch");
    assertThat(dispatchSpan.getAttributes().get(AttributeKey.stringKey("outbox.item.outcome")))
        .isEqualTo("skipped");
    assertThat(dispatchSpan.getAttributes().get(AttributeKey.stringKey("outbox.item.skip.reason")))
        .isEqualTo("already_sent");
    assertThat(spans).noneMatch(span -> span.getName().equals("outbox fcm send"));
    assertNoSensitiveTraceData(spans, notification);
  }

  @Test
  @DisplayName("TC4 - FCM 실패 시 실제 DB에 FAILED_RETRIABLE 저장 (retryCount+1, nextRetryAt 설정)")
  void shouldPersistFailedRetriableToDb_whenFcmThrows() throws FirebaseMessagingException {

    // Given: PENDING 이벤트 + FCM 예외 설정
    Notification notification = createAndSaveNotification(NotificationStatus.PENDING);
    OutboxEvent event = createAndSaveOutboxEvent(notification.getId());

    given(firebaseMessaging.send(any(Message.class))).willThrow(FirebaseMessagingException.class);
    resetSpans();

    // When: OutboxPublisher 실행

    Instant before = now();

    outboxPublisher.processPendingOutboxEvents();

    Instant after = now();

    // Then: OutboxEvent가 FAILED_RETRIABLE로 영속화되고 retryCount=1, nextRetryAt > now
    OutboxEvent processedEvent =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(processedEvent.getStatus()).isEqualTo(OutboxEventStatus.FAILED_RETRIABLE);
    assertThat(processedEvent.getRetryCount()).isEqualTo(1);
    assertThat(processedEvent.getNextRetryAt())
        .isBetween(before.plusSeconds(1), after.plusSeconds(1));

    List<SpanData> spans = awaitSpans(exporter, tracerProvider, 5);
    assertFailureTrace(spans, "retryable_failure", 1);
    assertThat(onlySpanNamed(spans, "outbox process").getStatus().getStatusCode())
        .isEqualTo(StatusCode.UNSET);
    assertThat(tracer.currentSpan()).isNull();
    assertNoSensitiveTraceData(spans, notification);
  }

  @Test
  @DisplayName("TC5 - retryCount=3에서 FCM 실패 시 실제 DB에 FAILED_PERMANENT 저장")
  void shouldPersistFailedPermanentToDb_whenRetryCountExceedsMax()
      throws FirebaseMessagingException {
    // Given
    Notification notification = createAndSaveNotification(NotificationStatus.PENDING);
    OutboxEvent event = createAndSaveOutboxEvent(notification.getId());
    event.setRetryCount(3); // 이미 3번 실패한 상태로 세팅
    outboxEventRepository.save(event);
    given(firebaseMessaging.send(any(Message.class))).willThrow(FirebaseMessagingException.class);
    resetSpans();

    // When
    outboxPublisher.processPendingOutboxEvents();

    // Then
    OutboxEvent saved =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(saved.getStatus()).isEqualTo(OutboxEventStatus.FAILED_PERMANENT);

    List<SpanData> spans = awaitSpans(exporter, tracerProvider, 5);
    assertFailureTrace(spans, "permanent_failure", 3);
    assertThat(onlySpanNamed(spans, "outbox process").getStatus().getStatusCode())
        .isEqualTo(StatusCode.UNSET);
  }

  @Test
  @DisplayName("TC6 - FAILED_RETRIABLE 이벤트가 nextRetryAt 경과 후 재폴링되어 SENT로 전이")
  void shouldRetryAndSendSuccessfully_whenNextRetryAtHasPassed() throws FirebaseMessagingException {
    // Given: Phase 1 - FCM 실패로 FAILED_RETRIABLE 상태로 전이
    Notification notification = createAndSaveNotification(NotificationStatus.PENDING);
    OutboxEvent event = createAndSaveOutboxEvent(notification.getId());
    given(firebaseMessaging.send(any(Message.class))).willThrow(FirebaseMessagingException.class);

    outboxPublisher.processPendingOutboxEvents(); // 1차 실행: FAILED_RETRIABLE로 전이

    // Given: Phase 2 - nextRetryAt을 과거로 조작하여 재폴링 조건 충족 (시간 경과 시뮬레이션)
    OutboxEvent failedEvent =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(failedEvent.getStatus()).isEqualTo(OutboxEventStatus.FAILED_RETRIABLE); // 중간 상태 확인

    failedEvent.setNextRetryAt(now().minusSeconds(10)); // 10초 전으로 조작 → 재폴링 대상
    outboxEventRepository.save(failedEvent);

    given(firebaseMessaging.send(any(Message.class))).willReturn("mock-message-id"); // FCM 성공으로 전환

    // When: 2차 실행 - FAILED_RETRIABLE 이벤트가 재폴링되어 처리
    outboxPublisher.processPendingOutboxEvents();

    // Then: 최종 상태 SENT 확인
    OutboxEvent saved =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(saved.getStatus()).isEqualTo(OutboxEventStatus.SENT);
    assertThat(saved.getProcessedAt()).isNotNull();
  }

  @Test
  @DisplayName("TC7 - FAILED_RETRIABLE이지만 재시도 시간이 지나지 않은 이벤트는 폴링 대상에서 제외")
  void shouldNotProcessRetriableEvent_whenNextRetryAtNotYetReached()
      throws FirebaseMessagingException {
    // Given: FCM 실패 → FAILED_RETRIABLE 전이 (nextRetryAt = now + 1초)
    Notification notification = createAndSaveNotification(NotificationStatus.PENDING);
    OutboxEvent event = createAndSaveOutboxEvent(notification.getId());
    given(firebaseMessaging.send(any(Message.class))).willThrow(FirebaseMessagingException.class);
    outboxPublisher.processPendingOutboxEvents();

    // Given: nextRetryAt이 아직 미래임을 확인
    OutboxEvent failedEvent =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(failedEvent.getStatus()).isEqualTo(OutboxEventStatus.FAILED_RETRIABLE);
    assertThat(failedEvent.getNextRetryAt()).isAfter(now()); // 아직 재시도 시간 미도달

    clearInvocations(firebaseMessaging);

    // When: 즉시 재폴링 (nextRetryAt 미도달 상태)
    outboxPublisher.processPendingOutboxEvents();

    // Then: FCM 호출 없음, 상태 변화 없음
    then(firebaseMessaging).should(never()).send(any(Message.class));
    OutboxEvent unchanged =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(unchanged.getStatus()).isEqualTo(OutboxEventStatus.FAILED_RETRIABLE);
    assertThat(unchanged.getRetryCount()).isEqualTo(1);
  }

  @Test
  @DisplayName("TC8 - 2회 연속 실패 후 nextRetryAt ≈ now+2s가 DB에 저장되는지 (backoff 증가 통합 검증)")
  void shouldPersistDoubledBackoff_whenFcmFailsTwiceConsecutively()
      throws FirebaseMessagingException {
    // Given: PENDING 이벤트 + FCM 계속 실패
    Notification notification = createAndSaveNotification(NotificationStatus.PENDING);
    OutboxEvent event = createAndSaveOutboxEvent(notification.getId());
    given(firebaseMessaging.send(any(Message.class))).willThrow(FirebaseMessagingException.class);

    // When: 1차 실패 (retryCount=0 → backoff 1s, nextRetryAt = now+1s)
    outboxPublisher.processPendingOutboxEvents();

    // Given: nextRetryAt을 과거로 조작하여 재폴링 조건 충족 (시간 경과 시뮬레이션)
    OutboxEvent afterFirst =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(afterFirst.getStatus()).isEqualTo(OutboxEventStatus.FAILED_RETRIABLE);
    assertThat(afterFirst.getRetryCount()).isEqualTo(1);

    afterFirst.setNextRetryAt(now().minusSeconds(10));
    outboxEventRepository.save(afterFirst);

    // When: 2차 실패 (retryCount=1 → backoff 2s, nextRetryAt = now+2s)
    Instant before = now();
    outboxPublisher.processPendingOutboxEvents();
    Instant after = now();

    // Then: retryCount=2, nextRetryAt ≈ now+2s가 실제 DB에 저장되는지 검증
    OutboxEvent afterSecond =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(afterSecond.getStatus()).isEqualTo(OutboxEventStatus.FAILED_RETRIABLE);
    assertThat(afterSecond.getRetryCount()).isEqualTo(2);
    assertThat(afterSecond.getNextRetryAt()).isBetween(before.plusSeconds(2), after.plusSeconds(2));
  }

  @Test
  @DisplayName("TC9 - UNREGISTERED 에러 코드 수신 시 retryCount=0에서도 즉시 FAILED_PERMANENT로 DB 저장")
  void shouldPersistFailedPermanent_whenFcmReturnsUnregistered() throws FirebaseMessagingException {
    // Given
    Notification notification = createAndSaveNotification(NotificationStatus.PENDING);
    OutboxEvent event = createAndSaveOutboxEvent(notification.getId());

    FirebaseMessagingException exception = mock(FirebaseMessagingException.class);
    given(exception.getMessagingErrorCode()).willReturn(MessagingErrorCode.UNREGISTERED);
    given(firebaseMessaging.send(any(Message.class))).willThrow(exception);
    resetSpans();

    // When
    outboxPublisher.processPendingOutboxEvents();

    // Then: Exponential Backoff 없이 즉시 영구 실패
    OutboxEvent saved =
        outboxEventRepository
            .findById(event.getId())
            .orElseThrow(() -> new IllegalStateException("OutboxEvent를 찾을 수 없습니다"));
    assertThat(saved.getStatus()).isEqualTo(OutboxEventStatus.FAILED_PERMANENT);
    assertThat(saved.getRetryCount()).isEqualTo(0);
    assertThat(saved.getProcessedAt()).isNotNull();

    List<SpanData> spans = awaitSpans(exporter, tracerProvider, 5);
    assertFailureTrace(spans, "permanent_failure", 0);
    assertThat(onlySpanNamed(spans, "outbox process").getStatus().getStatusCode())
        .isEqualTo(StatusCode.UNSET);
  }

  @Test
  @DisplayName("legacy null context도 Link 없는 consumer root span에서 처리")
  void shouldCreateUnlinkedConsumerRootForLegacyContext() {
    // Given
    Notification notification = createAndSaveNotification(NotificationStatus.PENDING);
    createAndSaveOutboxEvent(notification.getId());

    // When
    outboxPublisher.processPendingOutboxEvents();

    // Then
    SpanData processSpan = onlyProcessSpan(awaitSpans(exporter, tracerProvider, 1));
    assertThat(processSpan.getParentSpanContext().isValid()).isFalse();
    assertThat(processSpan.getLinks()).isEmpty();
  }

  @Test
  @DisplayName("재시도마다 새 trace를 만들고 동일 producer span을 Link")
  void shouldCreateIndependentLinkedTraceForEachRetry() throws FirebaseMessagingException {
    // Given
    Notification notification = createAndSaveNotification(NotificationStatus.PENDING);
    OutboxEvent event = createAndSaveOutboxEvent(notification.getId(), PRODUCER_TRACEPARENT);
    given(firebaseMessaging.send(any(Message.class))).willThrow(FirebaseMessagingException.class);

    // When: 첫 번째 처리 시도
    outboxPublisher.processPendingOutboxEvents();
    awaitSpans(exporter, tracerProvider, 1);
    OutboxEvent failedEvent = outboxEventRepository.findById(event.getId()).orElseThrow();
    failedEvent.setNextRetryAt(now().minusSeconds(10));
    outboxEventRepository.save(failedEvent);
    given(firebaseMessaging.send(any(Message.class))).willReturn("mock-message-id");

    // When: 두 번째 처리 시도
    outboxPublisher.processPendingOutboxEvents();

    // Then
    List<SpanData> processSpans = processSpans(awaitSpans(exporter, tracerProvider, 2));
    assertThat(processSpans).hasSize(2);
    assertThat(processSpans).extracting(SpanData::getTraceId).doesNotHaveDuplicates();
    assertThat(processSpans)
        .allSatisfy(
            span -> {
              assertThat(span.getKind()).isEqualTo(SpanKind.CONSUMER);
              assertThat(span.getParentSpanContext().isValid()).isFalse();
              assertThat(span.getLinks())
                  .singleElement()
                  .satisfies(
                      link -> {
                        assertThat(link.getSpanContext().getTraceId()).isEqualTo(PRODUCER_TRACE_ID);
                        assertThat(link.getSpanContext().getSpanId()).isEqualTo(PRODUCER_SPAN_ID);
                      });
            });
  }

  // TODO: 향후 Dead Letter Queue 도입 시 별도 테이블 이관 여부 검증 필요

  // TODO(후속 PR): createAndSaveNotification(status) 헬퍼를 상태별 메서드로 분리
  // - createAndSavePendingNotification() / createAndSaveSentNotification()
  // - status 파라미터 노출 시 sentAt 누락 등 fixture 불일치 실수 방지

  // ==========================
  // 테스트 헬퍼 메서드
  // ==========================

  /**
   * 랜덤 이메일을 가진 User 생성 및 저장
   *
   * <p>Why: Email unique constraint 회피를 위해 UUID 기반 랜덤 이메일 생성
   */
  private User createAndSaveUser() {
    String randomEmail = "test-" + UUID.randomUUID() + "@example.com";
    User user = User.create(randomEmail, "password123", "테스트유저");
    return userRepository.save(user);
  }

  private Notification createAndSaveNotification(NotificationStatus status) {
    // FK 제약 조건: User 먼저 생성
    User user = createAndSaveUser();

    Notification notification =
        Notification.createPending(
            user.getId(), // 실제 존재하는 User ID
            NotificationType.TEST,
            "테스트 알림",
            "테스트 메시지",
            "fcm-token-12345");
    notification.setStatus(status);

    if (status == NotificationStatus.SENT) {
      notification.setSentAt(now());
    }

    return notificationRepository.save(notification);
  }

  private OutboxEvent createAndSaveOutboxEvent(UUID notificationId) {
    return createAndSaveOutboxEvent(notificationId, null);
  }

  private OutboxEvent createAndSaveOutboxEvent(UUID notificationId, String producerTraceparent) {
    OutboxEvent event =
        OutboxEvent.createPending(
            notificationId, OutboxEventType.PUSH_NOTIFICATION, null, producerTraceparent);
    return outboxEventRepository.save(event);
  }

  private SpanData onlyProcessSpan(List<SpanData> spans) {
    List<SpanData> processSpans = processSpans(spans);
    assertThat(processSpans).hasSize(1);
    return processSpans.getFirst();
  }

  private SpanData onlySpanNamed(List<SpanData> spans, String name) {
    List<SpanData> namedSpans = spans.stream().filter(span -> span.getName().equals(name)).toList();
    assertThat(namedSpans).hasSize(1);
    return namedSpans.getFirst();
  }

  private void resetSpans() {
    flushSpans(tracerProvider);
    exporter.clear();
  }

  private void assertFailureTrace(List<SpanData> spans, String outcome, long retryCount) {
    SpanData dispatchSpan = onlySpanNamed(spans, "outbox dispatch");
    SpanData fcmSpan = onlySpanNamed(spans, "outbox fcm send");
    assertThat(dispatchSpan.getAttributes().get(AttributeKey.stringKey("outbox.item.outcome")))
        .isEqualTo(outcome);
    assertThat(dispatchSpan.getAttributes().get(AttributeKey.longKey("outbox.retry.count")))
        .isEqualTo(retryCount);
    assertThat(fcmSpan.getAttributes().get(AttributeKey.stringKey("outbox.fcm.outcome")))
        .isEqualTo(outcome);
  }

  private void assertNoSensitiveTraceData(List<SpanData> spans, Notification notification) {
    List<SpanData> runtimeSpans =
        spans.stream().filter(span -> span.getName().startsWith("outbox ")).toList();
    assertThat(runtimeSpans)
        .allSatisfy(
            span -> {
              String attributeKeys = span.getAttributes().asMap().keySet().toString();
              String attributeValues = span.getAttributes().asMap().values().toString();
              String eventValues = span.getEvents().toString();
              assertThat(attributeKeys)
                  .doesNotContainIgnoringCase("notificationId")
                  .doesNotContainIgnoringCase("outboxEventId")
                  .doesNotContainIgnoringCase("userId")
                  .doesNotContainIgnoringCase("token");
              assertThat(attributeValues)
                  .doesNotContain(notification.getId().toString())
                  .doesNotContain(notification.getRecipientToken());
              assertThat(eventValues)
                  .doesNotContain(notification.getId().toString())
                  .doesNotContain(notification.getRecipientToken());
            });
  }

  private List<SpanData> processSpans(List<SpanData> spans) {
    return spans.stream().filter(span -> span.getName().equals("outbox process")).toList();
  }
}