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

import dev.caskeleton.adapter.outbound.persistence.postgresql.inbox.PostgreSqlSameStoreInboxAdapter;
import dev.caskeleton.adapter.outbound.persistence.postgresql.outbox.PostgreSqlImmutableOutboxAppendAdapter;
import dev.caskeleton.adapter.outbound.persistence.postgresql.outbox.PostgreSqlPollingDeliveryAdapter;
import dev.caskeleton.application.inbox.*;
import dev.caskeleton.application.outbox.v2.*;
import dev.caskeleton.application.transaction.OperationId;
import java.time.*;
import org.flywaydb.core.Flyway;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;

public class PostgreSqlReplaySemanticProbe {
  public static void main(String[] args) {
    try (PostgreSqlReadinessSupport postgres = PostgreSqlReadinessSupport.start()) {
      migrate(postgres, "classpath:db/migration/postgresql", "flyway_schema_history");
      migrateIndependent(postgres, "classpath:db/migration/jpa/core", "flyway_jpa_core_history", "probe-core");
      migrateIndependent(postgres, "classpath:db/migration/jpa/inbox", "flyway_jpa_inbox_history", "probe-inbox");
      migrateIndependent(postgres, "classpath:db/migration/jpa/outbox-storage", "flyway_jpa_outbox_storage_history", "probe-outbox-storage");
      migrateIndependent(postgres, "classpath:db/migration/jpa/outbox-polling", "flyway_jpa_outbox_polling_history", "probe-outbox-polling");

      JdbcTemplate jdbc = new JdbcTemplate(postgres.dataSource());
      TransactionTemplate tx = new TransactionTemplate(new DataSourceTransactionManager(postgres.dataSource()));
      jdbc.update("update capability_schema_registry set lifecycle_state='ACTIVE' where capability_id in ('jpa-inbox-same-store-v1','jpa-outbox-storage-v2','jpa-outbox-polling-delivery-v2')");

      probeInboxForgedReplay(postgres, jdbc, tx);
      probeInboxRetention(postgres, jdbc, tx);
      probeOutboxRetrySchedule(postgres, jdbc, tx);
    }
  }

  private static void probeInboxForgedReplay(PostgreSqlReadinessSupport postgres, JdbcTemplate jdbc, TransactionTemplate tx) {
    PostgreSqlSameStoreInboxAdapter inbox = new PostgreSqlSameStoreInboxAdapter(postgres.dataSource());
    InboxScopeDigest scope = new InboxScopeDigest("c".repeat(64));
    InboxClaimAttempt attempt = inbox.newClaimAttempt(new OperationId("claim-owner-probe"));
    InboxOwner claimed = tx.execute(ignored -> ((InboxClaimOutcome.Acquired) inbox.claim(new InboxClaimRequest(scope, "d".repeat(64), attempt, Duration.ofSeconds(30), Duration.ofDays(7)))).owner());
    OperationId startOp = new OperationId("start-owner-probe");
    InboxOwner actual = tx.execute(ignored -> inbox.markProcessing(claimed, startOp).owner().orElseThrow());

    InboxOwner forged = new InboxOwner(scope, "forged-owner-token", actual.attempt(), actual.stateRevision(), actual.claimOperationId());
    InboxOwnerTransition replay = tx.execute(ignored -> inbox.markProcessing(forged, startOp));
    InboxOwner leaked = replay.owner().orElse(null);
    System.out.println("inboxForgedReplay.outcome=" + replay.outcome());
    System.out.println("inboxForgedReplay.returnedActualToken=" + (leaked != null && leaked.ownerToken().equals(actual.ownerToken())));
    System.out.println("inboxForgedReplay.returnedForgedToken=" + (leaked != null && leaked.ownerToken().equals(forged.ownerToken())));
    if (leaked != null) {
      var complete = tx.execute(ignored -> inbox.complete(leaked, new OperationId("complete-via-leaked-owner")));
      System.out.println("inboxForgedReplay.completeWithReturnedOwner=" + complete);
    }
  }

  private static void probeInboxRetention(PostgreSqlReadinessSupport postgres, JdbcTemplate jdbc, TransactionTemplate tx) {
    PostgreSqlSameStoreInboxAdapter inbox = new PostgreSqlSameStoreInboxAdapter(postgres.dataSource());
    InboxScopeDigest scope = new InboxScopeDigest("e".repeat(64));
    InboxClaimAttempt attempt = inbox.newClaimAttempt(new OperationId("claim-retention-probe"));
    InboxOwner claimed = tx.execute(ignored -> ((InboxClaimOutcome.Acquired) inbox.claim(new InboxClaimRequest(scope, "f".repeat(64), attempt, Duration.ofSeconds(30), Duration.ofDays(7)))).owner());
    InboxOwner processing = tx.execute(ignored -> inbox.markProcessing(claimed, new OperationId("start-retention-probe")).owner().orElseThrow());
    OperationId failOp = new OperationId("retry-retention-probe");
    var first = tx.execute(ignored -> inbox.markRetryable(processing, Duration.ofHours(1), failOp));
    var second = tx.execute(ignored -> inbox.markRetryable(processing, Duration.ofHours(9), failOp));
    Double remainingHours = jdbc.queryForObject("select extract(epoch from (retention_until - clock_timestamp())) / 3600.0 from inbox_record_v1 where scope_hash=?", Double.class, scope.value());
    System.out.println("inboxRetention.first=" + first);
    System.out.println("inboxRetention.secondDifferentRetention=" + second);
    System.out.printf("inboxRetention.remainingHours=%.3f%n", remainingHours == null ? -1.0 : remainingHours);
  }

  private static void probeOutboxRetrySchedule(PostgreSqlReadinessSupport postgres, JdbcTemplate jdbc, TransactionTemplate tx) {
    PostgreSqlImmutableOutboxAppendAdapter append = new PostgreSqlImmutableOutboxAppendAdapter(postgres.dataSource());
    PostgreSqlPollingDeliveryAdapter delivery = new PostgreSqlPollingDeliveryAdapter(postgres.dataSource());
    tx.executeWithoutResult(ignored -> cutover(jdbc));
    tx.executeWithoutResult(ignored -> append.append(new NewOutboxEventV2(
        "event-retry-semantic-probe", "WorkLog", "work-log-probe", 1, 0,
        "WorkLogChanged", 1, "portfolio.events", "work-log-probe", "application/json",
        "correlation-probe", null, Instant.parse("2026-08-29T00:00:00Z"), "{\"v\":1}")));
    ClaimedOutboxDelivery claimed = tx.execute(ignored -> delivery.claimBatch(new OutboxDeliveryClaimRequest("portfolio.events", "probe-relay", 1, Duration.ofSeconds(30)))).getFirst();
    OutboxDeliveryTransition transition = new OutboxDeliveryTransition(claimed.owner(), new OperationId("retry-schedule-probe"));
    Instant firstAt = Instant.now().plus(Duration.ofHours(1));
    Instant secondAt = firstAt.plus(Duration.ofHours(8));
    var first = tx.execute(ignored -> delivery.markRetryable(transition, firstAt, "BROKER.TEMP"));
    var second = tx.execute(ignored -> delivery.markRetryable(transition, secondAt, "BROKER.TEMP"));
    OffsetDateTime stored = jdbc.queryForObject("select next_attempt_at from outbox_delivery_v2 where retention_bucket=? and event_id=? and destination=?", OffsetDateTime.class, claimed.owner().retentionBucket(), claimed.owner().eventId(), claimed.owner().destination());
    System.out.println("outboxRetry.first=" + first);
    System.out.println("outboxRetry.secondDifferentSchedule=" + second);
    System.out.println("outboxRetry.storedEqualsFirst=" + (stored != null && Math.abs(Duration.between(firstAt, stored.toInstant()).toMillis()) < 10));
    System.out.println("outboxRetry.storedEqualsSecond=" + (stored != null && Math.abs(Duration.between(secondAt, stored.toInstant()).toMillis()) < 10));
  }

  private static void cutover(JdbcTemplate jdbc) {
    jdbc.queryForObject("select active_epoch from outbox_publication_control_v2 where scope_id='PRIMARY' for update", Long.class);
    jdbc.update("insert into outbox_publication_cutover_v2(scope_id,active_epoch,previous_epoch,transition_kind,active_authority,legacy_row_count,legacy_pending_count,legacy_digest,schema_manifest_id,external_manifest_id,activated_at) values ('PRIMARY',2,1,'CUTOVER','POLLING_V2',0,0,?,'jpa-outbox-storage-v2-schema-revision-2',null,clock_timestamp())", "0".repeat(64));
    jdbc.update("update outbox_publication_control_v2 set active_epoch=2,active_authority='POLLING_V2',revision=1,updated_at=clock_timestamp() where scope_id='PRIMARY' and active_epoch=1");
  }

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

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