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

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

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.UUID;
import org.flywaydb.core.Flyway;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.jdbc.core.JdbcTemplate;

/**
 * Real-PostgreSQL proof that the Fileserver metadata schema exists with the optimistic-locking and
 * READY-completeness guarantees the design requires.
 */
class PostgreSqlFileserverMigrationIntegrationTest {

  private static PostgreSqlReadinessSupport postgres;
  private static JdbcTemplate jdbc;

  @BeforeAll
  static void startAndMigratePostgreSql() {
    PostgreSqlReadinessSupport.assertDockerAvailable();
    postgres = PostgreSqlReadinessSupport.start();
    migrate("classpath:db/migration/postgresql", "flyway_schema_history");
    migrateIndependent(
        "classpath:db/migration/jpa/core", "flyway_jpa_core_history", "explicit-jpa-core-adoption");
    migrateIndependent(
        "classpath:db/migration/jpa/fileserver",
        "flyway_jpa_fileserver_history",
        "explicit-jpa-fileserver-adoption");
    jdbc = new JdbcTemplate(postgres.dataSource());
  }

  @AfterAll
  static void stopPostgreSql() {
    if (postgres != null) {
      postgres.close();
    }
  }

  @BeforeEach
  void clearRows() {
    jdbc.update("delete from fs_cleanup_item");
    jdbc.update("delete from fs_quota_reservation");
    jdbc.update("delete from fs_verification_result");
    jdbc.update("delete from fs_upload_session");
    jdbc.update("delete from fs_file");
  }

  @Test
  void createsFileserverTablesAndVersionColumns() throws Exception {
    try (Connection connection = postgres.dataSource().getConnection()) {
      assertThat(columnExists(connection, "fs_file", "version")).isTrue();
      assertThat(columnExists(connection, "fs_file", "content_key")).isTrue();
      assertThat(columnExists(connection, "fs_file", "strong_etag")).isTrue();
      assertThat(columnExists(connection, "fs_upload_session", "lease_until")).isTrue();
      assertThat(columnExists(connection, "fs_upload_session", "lease_token")).isTrue();
      assertThat(columnExists(connection, "fs_upload_session", "committed_offset")).isTrue();
      assertThat(columnExists(connection, "fs_quota_reservation", "reserved_bytes")).isTrue();
      assertThat(columnExists(connection, "fs_verification_result", "verdict")).isTrue();
      assertThat(columnExists(connection, "fs_cleanup_item", "next_attempt_at")).isTrue();
    }
  }

  @Test
  void registersTheCapabilityStreamAsInstalledButInactive() {
    String lifecycle =
        jdbc.queryForObject(
            "select lifecycle_state from capability_schema_registry where capability_id = ?",
            String.class,
            "jpa-fileserver-metadata-v1");

    assertThat(lifecycle).isEqualTo("INSTALLED_INACTIVE");
  }

  @Test
  void readyRowsMustCarryCompletePublishedIdentity() {
    UUID fileId = UUID.randomUUID();

    assertThatThrownBy(
            () ->
                jdbc.update(
                    """
                    insert into fs_file(
                        file_id, namespace, state, original_name, version, created_at, updated_at)
                    values (?, ?, 'READY', ?, 0, ?, ?)
                    """,
                    fileId,
                    "tenant-a",
                    "report.bin",
                    now(),
                    now()))
        .isInstanceOf(DataIntegrityViolationException.class);
  }

  @Test
  void contentKeyIsUniqueAcrossFilesButManyRowsMayHaveNone() {
    insertCreated(UUID.randomUUID());
    insertCreated(UUID.randomUUID());

    UUID first = UUID.randomUUID();
    UUID second = UUID.randomUUID();
    insertCreated(first);
    insertCreated(second);
    jdbc.update(
        "update fs_file set content_key = ? where file_id = ?", "ab/cd/key0000000001", first);

    assertThatThrownBy(
            () ->
                jdbc.update(
                    "update fs_file set content_key = ? where file_id = ?",
                    "ab/cd/key0000000001",
                    second))
        .isInstanceOf(DataIntegrityViolationException.class);
  }

  @Test
  void aLeaseIsAllOrNothing() {
    UUID fileId = UUID.randomUUID();
    UUID uploadId = UUID.randomUUID();
    insertCreated(fileId);
    insertUpload(uploadId, fileId);

    assertThatThrownBy(
            () ->
                jdbc.update(
                    "update fs_upload_session set lease_owner = ? where upload_id = ?",
                    "node-a",
                    uploadId))
        .isInstanceOf(DataIntegrityViolationException.class);
  }

  @Test
  void committedOffsetNeverExceedsTheDeclaredLength() {
    UUID fileId = UUID.randomUUID();
    UUID uploadId = UUID.randomUUID();
    insertCreated(fileId);
    insertUpload(uploadId, fileId);
    jdbc.update("update fs_upload_session set expected_length = 10 where upload_id = ?", uploadId);

    assertThatThrownBy(
            () ->
                jdbc.update(
                    "update fs_upload_session set committed_offset = 11 where upload_id = ?",
                    uploadId))
        .isInstanceOf(DataIntegrityViolationException.class);
  }

  private void insertCreated(UUID fileId) {
    jdbc.update(
        """
        insert into fs_file(
            file_id, namespace, state, original_name, version, created_at, updated_at)
        values (?, ?, 'CREATED', ?, 0, ?, ?)
        """,
        fileId,
        "tenant-a",
        "report.bin",
        now(),
        now());
  }

  private void insertUpload(UUID uploadId, UUID fileId) {
    jdbc.update(
        """
        insert into fs_upload_session(
            upload_id, file_id, protocol, committed_offset, expires_at, version,
            created_at, updated_at)
        values (?, ?, 'RAW', 0, ?, 0, ?, ?)
        """,
        uploadId,
        fileId,
        now().plusSeconds(3600),
        now(),
        now());
  }

  private static OffsetDateTime now() {
    return OffsetDateTime.now(ZoneOffset.UTC);
  }

  private static boolean columnExists(Connection connection, String table, String column)
      throws SQLException {
    try (ResultSet columns = connection.getMetaData().getColumns(null, null, table, column)) {
      return columns.next();
    }
  }

  private static void migrate(String location, String historyTable) {
    Flyway.configure()
        .dataSource(postgres.dataSource())
        .locations(location)
        .table(historyTable)
        .baselineOnMigrate(false)
        .outOfOrder(false)
        .load()
        .migrate();
  }

  private static void migrateIndependent(
      String location, String historyTable, String baselineDescription) {
    Flyway flyway =
        Flyway.configure()
            .dataSource(postgres.dataSource())
            .locations(location)
            .table(historyTable)
            .baselineVersion("0")
            .baselineDescription(baselineDescription)
            .baselineOnMigrate(false)
            .outOfOrder(false)
            .load();
    flyway.baseline();
    flyway.migrate();
  }

  @Test
  void analysisProbeRevisionTwoIsAcceptedWithoutRevisionThreeAndFourColumns() throws Exception {
    try (PostgreSqlReadinessSupport v2 = PostgreSqlReadinessSupport.start()) {
      Flyway.configure()
          .dataSource(v2.dataSource())
          .locations("classpath:db/migration/postgresql")
          .table("flyway_schema_history")
          .baselineOnMigrate(false)
          .outOfOrder(false)
          .load()
          .migrate();

      Flyway core =
          Flyway.configure()
              .dataSource(v2.dataSource())
              .locations("classpath:db/migration/jpa/core")
              .table("flyway_jpa_core_history")
              .baselineVersion("0")
              .baselineDescription("analysis-core")
              .baselineOnMigrate(false)
              .outOfOrder(false)
              .load();
      core.baseline();
      core.migrate();

      Flyway fileserver =
          Flyway.configure()
              .dataSource(v2.dataSource())
              .locations("classpath:db/migration/jpa/fileserver")
              .table("flyway_jpa_fileserver_history")
              .baselineVersion("0")
              .baselineDescription("analysis-fileserver-v2")
              .baselineOnMigrate(false)
              .outOfOrder(false)
              .target("2")
              .load();
      fileserver.baseline();
      fileserver.migrate();

      JdbcTemplate v2Jdbc = new JdbcTemplate(v2.dataSource());
      v2Jdbc.update(
          "update capability_schema_registry set lifecycle_state = 'ACTIVE' where capability_id = 'jpa-fileserver-metadata-v1'");
      new dev.caskeleton.adapter.outbound.persistence.fileserver.FileserverSchemaActivation(v2Jdbc)
          .requireActive();

      int revision =
          v2Jdbc.queryForObject(
              "select feature_revision from capability_schema_registry where capability_id = 'jpa-fileserver-metadata-v1'",
              Integer.class);
      boolean hasCleanupClaimToken;
      boolean hasUploadLifecycle;
      try (Connection connection = v2.dataSource().getConnection()) {
        hasCleanupClaimToken = columnExists(connection, "fs_cleanup_item", "claim_token");
        hasUploadLifecycle = columnExists(connection, "fs_upload_session", "lifecycle_state");
      }
      System.out.println("fileserverSchemaV2.featureRevision=" + revision);
      System.out.println("fileserverSchemaV2.activationAccepted=true");
      System.out.println("fileserverSchemaV2.cleanupClaimToken=" + hasCleanupClaimToken);
      System.out.println("fileserverSchemaV2.uploadLifecycleState=" + hasUploadLifecycle);
      assertThat(revision).isEqualTo(2);
      assertThat(hasCleanupClaimToken).isFalse();
      assertThat(hasUploadLifecycle).isFalse();
    }
  }

}