document-haness/docs/clean-architecture-backend-template/final/evidence/raw/071a-liveevent-full-sweep-probe.java

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

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import dev.caskeleton.adapter.outbound.persistence.platform.JpaPlatformContractSupport;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;

/**
 * The live-event log against a real PostgreSQL.
 *
 * <p>Every claim this store makes is about a constraint or about concurrency, and neither survives
 * a fake. A {@code CHECK} a fake does not enforce is a comment; two writers racing for the same
 * position has no meaning where only one connection exists — and that race is the whole reason the
 * position is part of the primary key rather than assigned in Java.
 */
@Tag("jpa-contract")
class LiveEventLogContractTest {

  private static final Instant NOW = Instant.parse("2026-08-25T09:00:00Z");
  private static final Duration RETENTION = Duration.ofHours(1);

  private static JpaPlatformContractSupport support;

  @BeforeAll
  static void startServer() {
    support = JpaPlatformContractSupport.start();
  }

  @AfterAll
  static void stopServer() {
    if (support != null) {
      support.close();
    }
  }

  @BeforeEach
  void migrate() throws SQLException, IOException {
    try (Connection connection = support.connection();
        Statement statement = connection.createStatement()) {
      statement.execute("DROP SCHEMA public CASCADE");
      statement.execute("CREATE SCHEMA public");
      statement.execute(migration());
    }
  }

  @Test
  @DisplayName("two positions cannot collide on one stream")
  void positionsAreUniquePerStream() throws SQLException {
    append("orders", 1, "a", NOW);

    assertThatThrownBy(() -> append("orders", 1, "b", NOW)).isInstanceOf(SQLException.class);
  }

  @Test
  @DisplayName("the same position on a different stream is fine")
  void positionsAreScopedToTheirStream() throws SQLException {
    append("orders", 1, "a", NOW);
    append("invoices", 1, "b", NOW);

    assertThat(replayAfter("orders", 0, NOW)).hasSize(1);
    assertThat(replayAfter("invoices", 0, NOW)).hasSize(1);
  }

  @Test
  @DisplayName("two writers racing for the same position produce one row, not two")
  void concurrentAppendsCannotShareAPosition() throws Exception {
    // The property the primary key exists for. Assigning positions in Java without it would let
    // both writers read the same maximum and both write it, and the second event would silently
    // overwrite the first — two different events at one address, with a client's cursor pointing
    // at whichever survived.
    List<String> outcomes = new ArrayList<>();
    try (Connection first = support.connection();
        Connection second = support.connection()) {
      first.setAutoCommit(false);
      second.setAutoCommit(false);
      insert(first, "orders", 1, "from-first", NOW);
      outcomes.add("first-inserted");
      first.commit();
      try {
        insert(second, "orders", 1, "from-second", NOW);
        second.commit();
        outcomes.add("second-inserted");
      } catch (SQLException refused) {
        second.rollback();
        outcomes.add("second-refused");
      }
    }

    assertThat(outcomes).containsExactly("first-inserted", "second-refused");
    assertThat(payloadAt("orders", 1)).isEqualTo("from-first");
  }

  @Test
  @DisplayName("a position of zero is refused by the database, not only by the code")
  void positionZeroIsRefused() {
    // A stream whose first row claimed 0 and one whose position was never assigned look identical
    // to a reader, so the floor is a constraint rather than a convention.
    assertThatThrownBy(() -> append("orders", 0, "a", NOW)).isInstanceOf(SQLException.class);
  }

  @Test
  @DisplayName("a row that expires before it happened is refused")
  void backwardsRetentionIsRefused() throws SQLException {
    try (Connection connection = support.connection();
        PreparedStatement statement =
            connection.prepareStatement(
                "INSERT INTO live_event_log (stream_id, position, payload, occurred_at,"
                    + " retained_until) VALUES (?, ?, ?, ?, ?)")) {
      statement.setString(1, "orders");
      statement.setLong(2, 1);
      statement.setString(3, "a");
      statement.setTimestamp(4, Timestamp.from(NOW));
      statement.setTimestamp(5, Timestamp.from(NOW.minusSeconds(1)));

      assertThatThrownBy(statement::executeUpdate).isInstanceOf(SQLException.class);
    }
  }

  @Test
  @DisplayName("a replay returns everything after a cursor, in order")
  void replayIsOrderedAndExclusive() throws SQLException {
    append("orders", 1, "a", NOW);
    append("orders", 2, "b", NOW);
    append("orders", 3, "c", NOW);

    assertThat(replayAfter("orders", 1, NOW)).containsExactly("b", "c");
    assertThat(replayAfter("orders", 3, NOW)).isEmpty();
  }

  @Test
  @DisplayName("an expired row is not replayed, even before it is swept")
  void expiredRowsAreNotReplayed() throws SQLException {
    append("orders", 1, "old", NOW.minus(Duration.ofHours(2)));
    append("orders", 2, "new", NOW);

    // Filtered on read as well as swept on a schedule. An expired row that has not been collected
    // yet is not history the client may have, and serving it would make the window this store
    // reports and the window it honours two different things.
    assertThat(replayAfter("orders", 0, NOW)).containsExactly("new");
    assertThat(earliestRetained("orders", NOW)).isEqualTo(2L);
  }

  @Test
  @DisplayName("a swept position is never handed out again")
  void sweptPositionsAreNotReused() throws SQLException {
    append("orders", 1, "gone", NOW.minus(Duration.ofHours(2)));
    append("orders", 2, "kept", NOW);
    assertThat(sweep(NOW)).isOne();

    // The highest-ever query deliberately ignores retention. Reusing position 1 would give two
    // different events the same address, and a client holding the older cursor would receive the
    // newer event as though it were the one it asked to continue after.
    assertThat(highestEverAssigned("orders")).isEqualTo(2L);
  }

  @Test
  @DisplayName("the sweep removes only what has expired")
  void sweepIsSelective() throws SQLException {
    append("orders", 1, "gone", NOW.minus(Duration.ofHours(2)));
    append("orders", 2, "kept", NOW);
    append("invoices", 1, "kept", NOW);

    assertThat(sweep(NOW)).isOne();
    assertThat(replayAfter("orders", 0, NOW)).containsExactly("kept");
    assertThat(replayAfter("invoices", 0, NOW)).containsExactly("kept");
  }

  @Test
  @DisplayName("an empty stream reports no window at all")
  void emptyStreamHasNoWindow() throws SQLException {
    assertThat(earliestRetained("nothing-here", NOW)).isNull();
  }

  @Test
  void analysisProbeFullySweptStreamReusesPosition() throws SQLException {
    append("orders", 1, "expired-only-row", NOW.minus(Duration.ofHours(2)));
    int swept = sweep(NOW);
    Long highestAfterFullSweep = highestEverAssigned("orders");
    long nextPosition = highestAfterFullSweep == null ? 1L : highestAfterFullSweep + 1L;
    append("orders", nextPosition, "new-event", NOW);
    System.out.println("liveEventFullSweep.swept=" + swept);
    System.out.println("liveEventFullSweep.highestAfterSweep=" + highestAfterFullSweep);
    System.out.println("liveEventFullSweep.nextPosition=" + nextPosition);
    System.out.println("liveEventFullSweep.payloadAtReusedPosition=" + payloadAt("orders", 1));
    assertThat(nextPosition).isEqualTo(1L);
  }

  private void append(String streamId, long position, String payload, Instant occurredAt)
      throws SQLException {
    try (Connection connection = support.connection()) {
      insert(connection, streamId, position, payload, occurredAt);
    }
  }

  private static void insert(
      Connection connection, String streamId, long position, String payload, Instant occurredAt)
      throws SQLException {
    try (PreparedStatement statement =
        connection.prepareStatement(
            "INSERT INTO live_event_log (stream_id, position, payload, occurred_at,"
                + " retained_until) VALUES (?, ?, ?, ?, ?)")) {
      statement.setString(1, streamId);
      statement.setLong(2, position);
      statement.setString(3, payload);
      statement.setTimestamp(4, Timestamp.from(occurredAt));
      statement.setTimestamp(5, Timestamp.from(occurredAt.plus(RETENTION)));
      statement.executeUpdate();
    }
  }

  private List<String> replayAfter(String streamId, long afterPosition, Instant now)
      throws SQLException {
    List<String> payloads = new ArrayList<>();
    try (Connection connection = support.connection();
        PreparedStatement statement =
            connection.prepareStatement(
                "SELECT payload FROM live_event_log WHERE stream_id = ? AND position > ?"
                    + " AND retained_until > ? ORDER BY position ASC")) {
      statement.setString(1, streamId);
      statement.setLong(2, afterPosition);
      statement.setTimestamp(3, Timestamp.from(now));
      try (ResultSet rows = statement.executeQuery()) {
        while (rows.next()) {
          payloads.add(rows.getString(1));
        }
      }
    }
    return payloads;
  }

  private Long earliestRetained(String streamId, Instant now) throws SQLException {
    try (Connection connection = support.connection();
        PreparedStatement statement =
            connection.prepareStatement(
                "SELECT min(position) FROM live_event_log WHERE stream_id = ?"
                    + " AND retained_until > ?")) {
      statement.setString(1, streamId);
      statement.setTimestamp(2, Timestamp.from(now));
      try (ResultSet rows = statement.executeQuery()) {
        rows.next();
        long value = rows.getLong(1);
        return rows.wasNull() ? null : value;
      }
    }
  }

  private Long highestEverAssigned(String streamId) throws SQLException {
    try (Connection connection = support.connection();
        PreparedStatement statement =
            connection.prepareStatement(
                "SELECT max(position) FROM live_event_log WHERE stream_id = ?")) {
      statement.setString(1, streamId);
      try (ResultSet rows = statement.executeQuery()) {
        rows.next();
        long value = rows.getLong(1);
        return rows.wasNull() ? null : value;
      }
    }
  }

  private String payloadAt(String streamId, long position) throws SQLException {
    try (Connection connection = support.connection();
        PreparedStatement statement =
            connection.prepareStatement(
                "SELECT payload FROM live_event_log WHERE stream_id = ? AND position = ?")) {
      statement.setString(1, streamId);
      statement.setLong(2, position);
      try (ResultSet rows = statement.executeQuery()) {
        return rows.next() ? rows.getString(1) : null;
      }
    }
  }

  private int sweep(Instant now) throws SQLException {
    try (Connection connection = support.connection();
        PreparedStatement statement =
            connection.prepareStatement("DELETE FROM live_event_log WHERE retained_until <= ?")) {
      statement.setTimestamp(1, Timestamp.from(now));
      return statement.executeUpdate();
    }
  }

  private static String migration() throws IOException {
    try (InputStream source =
        LiveEventLogContractTest.class.getResourceAsStream(
            "/db/migration/postgresql/V12__live_event_log.sql")) {
      if (source == null) {
        throw new IllegalStateException(
            "the migration this store is defined by is missing from the classpath, so the test"
                + " would create its own idea of the schema and certify that instead");
      }
      return new String(source.readAllBytes(), StandardCharsets.UTF_8);
    }
  }
}