package com.ankurm.expandcontract;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.jdbc.core.simple.JdbcClient;

import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

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

/**
 * The first of two failure modes covered in docs/14-the-ddl-lock-window.md, isolated here at
 * the raw JDBC level rather than through {@link com.ankurm.expandcontract.customer.CustomerService},
 * because that class's {@code withRetryForConcurrentDdl} would silently absorb exactly the
 * exception this test exists to show. While {@code ALTER TABLE ... DROP COLUMN} (V3) executes,
 * a concurrent, otherwise-correct statement against the same table can briefly see
 * {@code Table "CUSTOMERS" not found} - a real, transient condition, first noticed in this
 * module's own live load-generator run and reproduced here on demand. See
 * {@link DdlSilentDataLossTest} for the second, stranger failure mode the same root cause
 * produces - a statement that succeeds and is still lost, with no exception at all.
 */
class DdlCollisionExceptionTest {

    private static final int MAX_ATTEMPTS = 20;

    @Test
    void concurrentStatementCanSeeTableNotFoundWhileDropColumnRuns(@TempDir Path tmp) throws Exception {
        Transcript t = Transcript.start("14-ddl-collision-exception",
                "Failure mode 1: a concurrent statement that collides with DROP COLUMN, and is told so");

        for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
            Path db = tmp.resolve("collision-exception-" + attempt);
            TestSupport.migrateTo(db, "2");
            JdbcClient jdbc = TestSupport.jdbcClient(db);

            AtomicBoolean stop = new AtomicBoolean(false);
            List<String> collisionMessages = new CopyOnWriteArrayList<>();
            List<Long> okCount = new CopyOnWriteArrayList<>();
            CountDownLatch started = new CountDownLatch(1);

            Thread reader = new Thread(() -> {
                started.countDown();
                long ok = 0;
                while (!stop.get()) {
                    try {
                        jdbc.sql("SELECT COUNT(*) FROM customers").query(Long.class).single();
                        ok++;
                    } catch (Exception ex) {
                        Throwable root = ex;
                        while (root.getCause() != null) {
                            root = root.getCause();
                        }
                        if (root.getMessage() != null && root.getMessage().contains("CUSTOMERS")) {
                            collisionMessages.add(root.getClass().getName() + ": " + root.getMessage());
                        }
                    }
                }
                okCount.add(ok);
            });
            reader.start();
            started.await();

            // Deploy 4b, running live while the reader above keeps querying the same table.
            TestSupport.migrateTo(db, "3");

            stop.set(true);
            reader.join();

            if (collisionMessages.isEmpty() && attempt < MAX_ATTEMPTS) {
                continue;
            }

            t.line("attempts needed to reproduce the race: " + attempt + " of " + MAX_ATTEMPTS);
            t.line("successful reads while DROP COLUMN was in flight: " + okCount.get(0));
            t.line("reads that collided with the in-flight DROP COLUMN: " + collisionMessages.size());
            if (!collisionMessages.isEmpty()) {
                t.line("example: " + collisionMessages.get(0));
            }
            t.write();

            assertThat(collisionMessages)
                    .withFailMessage("expected at least one concurrent read to collide with DROP COLUMN "
                            + "within %d attempts", MAX_ATTEMPTS)
                    .isNotEmpty();
            return;
        }
    }
}