package com.easyagents.scheduler.quartz;

import com.easyagents.scheduler.ConcurrencyPolicy;
import com.easyagents.scheduler.MisfirePolicy;
import com.easyagents.scheduler.OnceSchedulePlan;
import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleFireContext;
import com.easyagents.scheduler.ScheduleHandler;
import com.easyagents.scheduler.ScheduleId;
import com.easyagents.scheduler.ScheduleRefireException;
import org.h2.jdbcx.JdbcDataSource;
import org.h2.tools.RunScript;
import org.junit.Test;

import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.time.Instant;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/**
 * Quartz JDBC JobStore、建表资源和外部 DataSource 所有权的集成测试。
 */
public class QuartzJdbcIntegrationTest {

    /**
     * 验证 H2 建表脚本、结构校验和 JDBC JobStore 实际触发链路。
     */
    @Test
    public void shouldRunWithCallerManagedDataSource() throws Exception {
        JdbcDataSource dataSource = dataSource();
        executeSchema(dataSource);
        QuartzSchemaValidator.validate(dataSource, "QRTZ_");

        CountDownLatch latch = new CountDownLatch(1);
        QuartzSchedulerConfig config = jdbcConfig();
        QuartzScheduleService service = QuartzSchedulerFactory.createJdbc(
            dataSource,
            config,
            List.of(handler(latch)),
            List.of()
        );
        try {
            service.start();
            service.create(new ScheduleDefinition(
                new ScheduleId("jdbc", "once"),
                "jdbc-handler",
                new OnceSchedulePlan(Instant.now().plusMillis(500)),
                MisfirePolicy.FIRE_ONCE_NOW,
                ConcurrencyPolicy.DISALLOW,
                true,
                Map.of("source", "h2"),
                "jdbc integration"
            ));
            assertTrue("JDBC handler did not execute", latch.await(8, TimeUnit.SECONDS));
        } finally {
            service.close();
        }

        // 同一 JVM 使用相同 Scheduler Name 重建时覆盖已释放的内部 Provider 引用。
        CountDownLatch restartedLatch = new CountDownLatch(1);
        QuartzScheduleService restarted = QuartzSchedulerFactory.createJdbc(
            dataSource,
            config,
            List.of(handler(restartedLatch)),
            List.of()
        );
        try {
            restarted.start();
            assertTrue(restarted.get(new ScheduleId("jdbc", "once")).isPresent());
            restarted.triggerNow(
                new ScheduleId("jdbc", "once"),
                "restart-trigger",
                Map.of()
            );
            assertTrue(
                "rebuilt scheduler did not trigger persisted job",
                restartedLatch.await(5, TimeUnit.SECONDS)
            );
        } finally {
            restarted.close();
        }

        // Provider 只释放内部强引用，调用方 DataSource 仍可继续使用。
        try (
            Connection connection = dataSource.getConnection();
            Statement statement = connection.createStatement();
            ResultSet resultSet = statement.executeQuery("SELECT COUNT(*) FROM QRTZ_JOB_DETAILS")
        ) {
            assertTrue(resultSet.next());
            assertEquals(1, resultSet.getInt(1));
        }
    }

    /**
     * 验证缺少 Quartz 表时结构校验拒绝启动。
     */
    @Test
    public void shouldRejectMissingSchema() {
        try {
            QuartzSchemaValidator.validate(dataSource(), "QRTZ_");
            fail("expected schema validation failure");
        } catch (ScheduleException exception) {
            assertEquals(ScheduleErrorCode.SCHEMA_INVALID, exception.errorCode());
        }
    }

    /**
     * 验证关键列缺失时结构校验不会把不兼容结构误判为可用。
     */
    @Test
    public void shouldRejectMissingCriticalColumn() throws Exception {
        JdbcDataSource dataSource = dataSource();
        executeSchema(dataSource);
        try (
            Connection connection = dataSource.getConnection();
            Statement statement = connection.createStatement()
        ) {
            statement.execute("ALTER TABLE QRTZ_CRON_TRIGGERS DROP COLUMN CRON_EXPRESSION");
        }

        try {
            QuartzSchemaValidator.validate(dataSource, "QRTZ_");
            fail("expected critical column validation failure");
        } catch (ScheduleException exception) {
            assertEquals(ScheduleErrorCode.SCHEMA_INVALID, exception.errorCode());
            assertTrue(exception.getMessage().contains("CRON_EXPRESSION"));
        }
    }

    /**
     * 验证未限定前缀只校验当前 Schema，不会接受其他 Schema 的同名表。
     */
    @Test
    public void shouldRejectQuartzTablesOutsideCurrentSchema() throws Exception {
        JdbcDataSource dataSource = dataSource();
        try (
            Connection connection = dataSource.getConnection();
            Statement statement = connection.createStatement()
        ) {
            statement.execute("CREATE SCHEMA OTHER");
            statement.execute("SET SCHEMA OTHER");
            executeSchema(connection);
        }

        try {
            QuartzSchemaValidator.validate(dataSource, "QRTZ_");
            fail("expected current schema validation failure");
        } catch (ScheduleException exception) {
            assertEquals(ScheduleErrorCode.SCHEMA_INVALID, exception.errorCode());
        }
        QuartzSchemaValidator.validate(dataSource, "OTHER.QRTZ_");
    }

    /**
     * 验证 JDBC JobStore 能原子删除仍在执行立即触发的任务。
     */
    @Test
    public void shouldDeleteRunningScheduleAtomicallyInJdbcStore() throws Exception {
        JdbcDataSource dataSource = dataSource();
        executeSchema(dataSource);
        CountDownLatch started = new CountDownLatch(1);
        CountDownLatch release = new CountDownLatch(1);
        CountDownLatch completed = new CountDownLatch(1);
        ScheduleHandler blockingHandler = new ScheduleHandler() {
            @Override
            public String code() {
                return "delete-handler";
            }

            @Override
            public void execute(ScheduleFireContext context) throws InterruptedException {
                started.countDown();
                release.await();
                completed.countDown();
            }
        };
        QuartzScheduleService service = QuartzSchedulerFactory.createJdbc(
            dataSource,
            jdbcConfig(),
            List.of(blockingHandler),
            List.of()
        );
        ScheduleId scheduleId = new ScheduleId("jdbc", "delete-running");
        try {
            service.start();
            service.create(new ScheduleDefinition(
                scheduleId,
                blockingHandler.code(),
                new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
                MisfirePolicy.FIRE_ONCE_NOW,
                ConcurrencyPolicy.DISALLOW,
                true,
                Map.of(),
                "delete running"
            ));
            service.triggerNow(scheduleId, "delete-running-1", Map.of());
            assertTrue("JDBC handler did not start", started.await(5, TimeUnit.SECONDS));

            assertTrue(service.delete(scheduleId));
            assertTrue(service.get(scheduleId).isEmpty());
        } finally {
            release.countDown();
            completed.await(5, TimeUnit.SECONDS);
            service.close();
        }
    }

    /**
     * 验证延迟重试先落入 JDBC JobStore，关闭并重建节点后仍使用原 fire 上下文执行。
     */
    @Test
    public void shouldRecoverPersistedRetryAfterSchedulerRestart() throws Exception {
        JdbcDataSource dataSource = dataSource();
        executeSchema(dataSource);
        QuartzSchedulerConfig config = jdbcConfig();
        CountDownLatch firstFailed = new CountDownLatch(1);
        CountDownLatch retryCompleted = new CountDownLatch(1);
        CopyOnWriteArrayList<ScheduleFireContext> contexts = new CopyOnWriteArrayList<>();
        ScheduleHandler retryingHandler = new ScheduleHandler() {
            @Override
            public String code() {
                return "persistent-retry-handler";
            }

            @Override
            public void execute(ScheduleFireContext context) {
                contexts.add(context);
                if (contexts.size() == 1) {
                    firstFailed.countDown();
                    throw new ScheduleRefireException("temporary database failure",
                        new IllegalStateException("unavailable"), 2,
                        Duration.ofSeconds(2));
                }
                retryCompleted.countDown();
            }
        };
        ScheduleId scheduleId = new ScheduleId("jdbc", "persistent-retry");
        QuartzScheduleService first = QuartzSchedulerFactory.createJdbc(
            dataSource, config, List.of(retryingHandler), List.of());
        first.start();
        first.create(new ScheduleDefinition(
            scheduleId,
            retryingHandler.code(),
            new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
            MisfirePolicy.FIRE_ONCE_NOW,
            ConcurrencyPolicy.DISALLOW,
            true,
            Map.of("snapshot", "original"),
            "persistent retry"
        ));
        first.triggerNow(scheduleId, "persistent-invocation", Map.of());
        assertTrue(firstFailed.await(5, TimeUnit.SECONDS));
        awaitRetryTrigger(dataSource);
        first.close();

        QuartzScheduleService restarted = QuartzSchedulerFactory.createJdbc(
            dataSource, config, List.of(retryingHandler), List.of());
        try {
            restarted.start();
            assertTrue("persisted retry did not execute after restart",
                retryCompleted.await(8, TimeUnit.SECONDS));
            assertEquals(contexts.get(0).scheduledFireTime(), contexts.get(1).scheduledFireTime());
            assertEquals(contexts.get(0).actualFireTime(), contexts.get(1).actualFireTime());
            assertEquals(contexts.get(0).fireInstanceId(), contexts.get(1).fireInstanceId());
            assertEquals("persistent-invocation", contexts.get(1).invocationId());
        } finally {
            restarted.close();
        }
    }

    private static void awaitRetryTrigger(JdbcDataSource dataSource) throws Exception {
        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
        do {
            try (Connection connection = dataSource.getConnection();
                 Statement statement = connection.createStatement();
                 ResultSet result = statement.executeQuery(
                     "SELECT COUNT(*) FROM QRTZ_TRIGGERS WHERE TRIGGER_GROUP LIKE 'ea.scheduler.retry.%'")) {
                if (result.next() && result.getInt(1) > 0) return;
            }
            Thread.sleep(25L);
        } while (System.nanoTime() < deadline);
        fail("persistent retry trigger was not stored");
    }

    private static JdbcDataSource dataSource() {
        JdbcDataSource dataSource = new JdbcDataSource();
        dataSource.setURL("jdbc:h2:mem:scheduler-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
        dataSource.setUser("sa");
        dataSource.setPassword("");
        return dataSource;
    }

    private static void executeSchema(JdbcDataSource dataSource) throws Exception {
        try (Connection connection = dataSource.getConnection()) {
            executeSchema(connection);
        }
    }

    /**
     * 在指定连接的当前 Schema 中执行 H2 Quartz 建表资源。
     *
     * @param connection 目标数据库连接
     * @throws Exception 资源读取或脚本执行失败时抛出
     */
    private static void executeSchema(Connection connection) throws Exception {
        InputStream stream = QuartzJdbcIntegrationTest.class
            .getClassLoader()
            .getResourceAsStream("quartz-schema/h2-2.5.2.sql");
        if (stream == null) {
            throw new IllegalStateException("H2 Quartz schema resource not found");
        }
        try (InputStreamReader reader = new InputStreamReader(stream, StandardCharsets.UTF_8)) {
            RunScript.execute(connection, reader);
        }
    }

    private static QuartzSchedulerConfig jdbcConfig() {
        return new QuartzSchedulerConfig(
            "jdbc-scheduler-" + UUID.randomUUID(),
            "NON_CLUSTERED",
            "QRTZ_",
            QuartzSchedulerConfig.STANDARD_JDBC_DELEGATE,
            false,
            2,
            Thread.NORM_PRIORITY,
            1,
            0L,
            15_000L,
            1_000L,
            true,
            30_000L,
            true
        );
    }

    private static ScheduleHandler handler(CountDownLatch latch) {
        return new ScheduleHandler() {
            @Override
            public String code() {
                return "jdbc-handler";
            }

            @Override
            public void execute(ScheduleFireContext context) {
                latch.countDown();
            }
        };
    }
}
