document-haness/docs/clean-architecture-backend-template/final/evidence/raw/075a-outbox-stale-worker-state-regression-probe.java

package dev.caskeleton.adapter.outbound.persistence.outbox;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentCaptor.forClass;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import dev.caskeleton.adapter.outbound.persistence.outbox.entity.OutboxEventEntity;
import dev.caskeleton.application.outbox.NewOutboxEvent;
import dev.caskeleton.application.outbox.OutboxEvent;
import dev.caskeleton.application.outbox.OutboxEventStatus;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;

/**
 * Unit tests for {@link OutboxStoreAdapter} — covers append mapping, claim transition, mark
 * operations, countByStatus, and oldestUnpublishedAgeSecondsByEventType.
 *
 * <h2>FIFO gate location contract</h2>
 *
 * <p>The per-aggregate FIFO gate (I4) lives entirely in the {@link
 * OutboxClaimRepository#claimEligible} native SQL query via a {@code NOT EXISTS} correlated
 * subquery. The adapter itself performs NO in-memory filtering: every row returned by the claim
 * repository passes through to the caller. Verifying that the SQL gate enforces FIFO is the
 * responsibility of the Testcontainers contract test in {@code app-bootstrap}.
 *
 * <p>Real SKIP LOCKED / FIFO gate behaviour is covered by the Task E PG contract tests.
 */
class OutboxStoreAdapterTest {

  private static final Instant NOW = Instant.parse("2026-06-11T10:00:00Z");
  private static final Duration IN_FLIGHT_TIMEOUT = Duration.ofMinutes(5);

  private final OutboxEventJpaRepository repo = mock(OutboxEventJpaRepository.class);
  private final OutboxClaimRepository claimRepo = mock(OutboxClaimRepository.class);
  private final OutboxStoreAdapter adapter = new OutboxStoreAdapter(repo, claimRepo);

  // ---- helper ----

  private static NewOutboxEvent newEvent(String eventId) {
    return new NewOutboxEvent(eventId, "UserCreated", "agg-1", "{}", NOW, "corr-1", "idem-1");
  }

  private static OutboxEventEntity pendingEntity(String eventId) {
    OutboxEventEntity e = new OutboxEventEntity();
    e.setEventId(eventId);
    e.setAggregateId("agg-1");
    e.setEventType("UserCreated");
    e.setPayload("{}");
    e.setOccurredAt(NOW.minusSeconds(60));
    e.setStatus("PENDING");
    e.setAttemptCount(0);
    e.setNextAttemptAt(NOW.minusSeconds(60));
    e.setCorrelationId("corr-1");
    e.setIdempotencyKey("idem-1");
    return e;
  }

  // ---- append ----

  @Test
  void appendPersistsEntityWithCorrectFieldMapping() {
    NewOutboxEvent event = newEvent("evt-001");

    adapter.append(event);

    ArgumentCaptor<OutboxEventEntity> saved = forClass(OutboxEventEntity.class);
    verify(repo).save(saved.capture());
    OutboxEventEntity entity = saved.getValue();

    assertThat(entity.getEventId()).isEqualTo("evt-001");
    assertThat(entity.getEventType()).isEqualTo("UserCreated");
    assertThat(entity.getAggregateId()).isEqualTo("agg-1");
    assertThat(entity.getPayload()).isEqualTo("{}");
    assertThat(entity.getOccurredAt()).isEqualTo(NOW);
    assertThat(entity.getStatus()).isEqualTo("PENDING");
    assertThat(entity.getAttemptCount()).isEqualTo(0);
    assertThat(entity.getNextAttemptAt()).isEqualTo(NOW); // nextAttemptAt = occurredAt on PENDING
    assertThat(entity.getCorrelationId()).isEqualTo("corr-1");
    assertThat(entity.getIdempotencyKey()).isEqualTo("idem-1");
  }

  // ---- claimBatch ----

  @Test
  void claimBatchTransitionsEntityToInFlightAndReturnsMappedEvent() {
    OutboxEventEntity entity = pendingEntity("evt-002");
    when(claimRepo.claimEligible(eq(NOW), eq(2))).thenReturn(List.of(entity));

    List<OutboxEvent> claimed = adapter.claimBatch(2, NOW, IN_FLIGHT_TIMEOUT);

    // status and attemptCount updated on entity
    assertThat(entity.getStatus()).isEqualTo("IN_FLIGHT");
    assertThat(entity.getAttemptCount()).isEqualTo(1);
    assertThat(entity.getNextAttemptAt()).isEqualTo(NOW.plus(IN_FLIGHT_TIMEOUT));

    // returned OutboxEvent reflects post-transition state
    assertThat(claimed).hasSize(1);
    OutboxEvent result = claimed.get(0);
    assertThat(result.eventId()).isEqualTo("evt-002");
    assertThat(result.status()).isEqualTo(OutboxEventStatus.IN_FLIGHT);
    assertThat(result.attemptCount()).isEqualTo(1);
  }

  @Test
  void claimBatchReturnsEmptyWhenRepoReturnsNothing() {
    when(claimRepo.claimEligible(any(), anyInt())).thenReturn(List.of());

    assertThat(adapter.claimBatch(5, NOW, IN_FLIGHT_TIMEOUT)).isEmpty();
  }

  /**
   * Regression guard: the per-aggregate FIFO gate lives in the SQL query (I4). The adapter must NOT
   * apply any in-memory filtering — all rows returned by the repository must appear in the result,
   * including multiple rows for the same aggregate.
   *
   * <p>If this test fails it means in-memory filtering was re-introduced in the adapter; the fix is
   * to move that logic back to the {@code claimEligible} SQL query.
   */
  @Test
  void claimBatchPassesAllRepoResultsThroughWithoutInMemoryFifoFiltering() {
    // Two rows for the same aggregate — repo returns both (SQL gate already enforces FIFO)
    OutboxEventEntity head = pendingEntity("evt-head");
    OutboxEventEntity tail = new OutboxEventEntity();
    tail.setEventId("evt-tail");
    tail.setAggregateId("agg-1"); // same aggregate as head
    tail.setEventType("UserCreated");
    tail.setPayload("{}");
    tail.setOccurredAt(NOW.minusSeconds(30)); // later than head (head is at -60s)
    tail.setStatus("PENDING");
    tail.setAttemptCount(0);
    tail.setNextAttemptAt(NOW.minusSeconds(30));
    tail.setCorrelationId("corr-2");
    tail.setIdempotencyKey("idem-2");

    when(claimRepo.claimEligible(eq(NOW), eq(10))).thenReturn(List.of(head, tail));

    List<OutboxEvent> claimed = adapter.claimBatch(10, NOW, IN_FLIGHT_TIMEOUT);

    // Adapter must return both rows — no in-memory FIFO filtering
    assertThat(claimed)
        .hasSize(2)
        .extracting(OutboxEvent::eventId)
        .containsExactly("evt-head", "evt-tail");
  }

  // ---- markPublished ----

  @Test
  void markPublishedUpdatesStatusToPublished() {
    OutboxEventEntity entity = pendingEntity("evt-003");
    entity.setStatus("IN_FLIGHT");
    when(repo.findById("evt-003")).thenReturn(Optional.of(entity));

    adapter.markPublished("evt-003");

    assertThat(entity.getStatus()).isEqualTo("PUBLISHED");
  }

  // ---- markFailed ----

  @Test
  void markFailedSetsFailedStatusAndNextAttemptAt() {
    OutboxEventEntity entity = pendingEntity("evt-004");
    entity.setStatus("IN_FLIGHT");
    when(repo.findById("evt-004")).thenReturn(Optional.of(entity));
    Instant retryAt = NOW.plus(Duration.ofSeconds(30));

    adapter.markFailed("evt-004", retryAt);

    assertThat(entity.getStatus()).isEqualTo("FAILED");
    assertThat(entity.getNextAttemptAt()).isEqualTo(retryAt);
  }

  // ---- markDead ----

  @Test
  void markDeadSetsDeadStatus() {
    OutboxEventEntity entity = pendingEntity("evt-005");
    entity.setStatus("IN_FLIGHT");
    when(repo.findById("evt-005")).thenReturn(Optional.of(entity));

    adapter.markDead("evt-005");

    assertThat(entity.getStatus()).isEqualTo("DEAD");
  }

  // ---- countByStatus ----

  @Test
  void countByStatusMapsProjectionResultsToStatusEnumMap() {
    List<Object[]> rows =
        List.of(
            new Object[] {"PENDING", 5L},
            new Object[] {"IN_FLIGHT", 2L},
            new Object[] {"PUBLISHED", 10L});
    when(repo.countGroupedByStatus()).thenReturn(rows);

    Map<OutboxEventStatus, Long> counts = adapter.countByStatus();

    assertThat(counts).containsEntry(OutboxEventStatus.PENDING, 5L);
    assertThat(counts).containsEntry(OutboxEventStatus.IN_FLIGHT, 2L);
    assertThat(counts).containsEntry(OutboxEventStatus.PUBLISHED, 10L);
    assertThat(counts).doesNotContainKey(OutboxEventStatus.FAILED);
    assertThat(counts).doesNotContainKey(OutboxEventStatus.DEAD);
  }

  // ---- oldestUnpublishedAgeSecondsByEventType ----

  @Test
  void oldestUnpublishedAgeComputesSecondsFromOldestRowOccurredAt() {
    // oldest row for "UserCreated" occurred 120 seconds before NOW
    Instant oldestAt = NOW.minusSeconds(120);
    List<Object[]> rows = List.<Object[]>of(new Object[] {"UserCreated", oldestAt});
    when(repo.findOldestUnpublishedOccurredAtByEventType()).thenReturn(rows);

    Map<String, Long> ages = adapter.oldestUnpublishedAgeSecondsByEventType(NOW);

    assertThat(ages).containsEntry("UserCreated", 120L);
  }

  @Test
  void oldestUnpublishedAgeReturnsEmptyMapWhenNoUnpublishedRows() {
    when(repo.findOldestUnpublishedOccurredAtByEventType()).thenReturn(List.of());

    assertThat(adapter.oldestUnpublishedAgeSecondsByEventType(NOW)).isEmpty();
  }

  // ---- missing-entity guard (fail-closed consistency) ----

  /**
   * When the outbox row is not found during markPublished, the adapter must throw {@link
   * IllegalStateException} with the eventId in the message instead of silently doing nothing. A
   * silent no-op leaves the row IN_FLIGHT forever, blocking the aggregate's FIFO queue with no
   * error observable by the relay.
   */
  @Test
  void markPublishedThrowsWhenEventIdNotFound() {
    when(repo.findById("missing-evt")).thenReturn(Optional.empty());

    assertThatThrownBy(() -> adapter.markPublished("missing-evt"))
        .isInstanceOf(IllegalStateException.class)
        .hasMessageContaining("missing-evt");
  }

  /**
   * When the outbox row is not found during markFailed, the adapter must throw {@link
   * IllegalStateException} with the eventId in the message. A silent no-op would leave the relay
   * believing a retry is scheduled while the row remains IN_FLIGHT, blocking the aggregate queue
   * and producing no error log.
   */
  @Test
  void markFailedThrowsWhenEventIdNotFound() {
    when(repo.findById("missing-evt")).thenReturn(Optional.empty());
    Instant retryAt = NOW.plus(Duration.ofSeconds(30));

    assertThatThrownBy(() -> adapter.markFailed("missing-evt", retryAt))
        .isInstanceOf(IllegalStateException.class)
        .hasMessageContaining("missing-evt");
  }

  /**
   * When the outbox row is not found during markDead, the adapter must throw {@link
   * IllegalStateException} with the eventId in the message. A silent no-op would leave the row
   * IN_FLIGHT instead of DEAD, preventing runbook-level visibility and manual resolution.
   */
  @Test
  void markDeadThrowsWhenEventIdNotFound() {
    when(repo.findById("missing-evt")).thenReturn(Optional.empty());

    assertThatThrownBy(() -> adapter.markDead("missing-evt"))
        .isInstanceOf(IllegalStateException.class)
        .hasMessageContaining("missing-evt");
  }
  @Test
  void analysisProbeStaleWorkerCanRegressPublishedRowToFailed() {
    OutboxEventEntity entity = pendingEntity("evt-stale");
    entity.setStatus("PUBLISHED");
    when(repo.findById("evt-stale")).thenReturn(Optional.of(entity));
    Instant retryAt = NOW.plus(Duration.ofSeconds(30));

    adapter.markFailed("evt-stale", retryAt);

    System.out.println("outboxStaleWorker.before=PUBLISHED");
    System.out.println("outboxStaleWorker.after=" + entity.getStatus());
    System.out.println("outboxStaleWorker.retryAt=" + entity.getNextAttemptAt());
    assertThat(entity.getStatus()).isEqualTo("FAILED");
  }

}