package study.batchquartz;

import org.junit.jupiter.api.Test;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.Trigger;
import org.quartz.TriggerKey;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

/// U46 判定实验之一：Quartz 持久调度、集群互斥、misfire、故障接管与“调度成功≠业务成功”。
/// 所有断言都落在真实 MySQL 的 QRTZ 表与真实 Quartz 2.5.2 调度器上。
class QuartzPersistenceTest {

    @Test
    void jdbcStorePersistsAcrossSchedulerRestart() throws Exception {
        try (LabDatabase db = new LabDatabase("org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql")) {
            QuartzFixture fixture = new QuartzFixture(db);
            Scheduler first = fixture.scheduler("u46-a", false, 7500);
            JobKey key = fixture.schedule(first, "persisted", QuartzFixture.RecordingJob.class,
                    QuartzFixture.onceIn(60_000, "persisted-trigger"), false);
            assertEquals("WAITING", state(db, "persisted-trigger"));
            first.shutdown(false);

            Scheduler second = fixture.scheduler("u46-b", false, 7500);
            assertNotNull(second.getJobDetail(key), "重启后 JobDetail 仍在 QRTZ_JOB_DETAILS");
            Trigger trigger = second.getTrigger(TriggerKey.triggerKey("persisted-trigger", "u46"));
            assertNotNull(trigger);
            assertTrue(trigger.getNextFireTime().getTime() > System.currentTimeMillis(),
                    "下次触发时间也是持久状态");
            assertEquals("WAITING", state(db, "persisted-trigger"));
            assertEquals(1L, db.count("SELECT COUNT(*) FROM QRTZ_JOB_DETAILS WHERE SCHED_NAME=? AND JOB_NAME=?",
                    QuartzFixture.SCHED_NAME, "persisted"));
            second.shutdown(false);
            System.out.println("U46 quartzPersisted jobRows=" + db.count(
                    "SELECT COUNT(*) FROM QRTZ_JOB_DETAILS") + " triggerState=" + state(db, "persisted-trigger"));
        }
    }

    @Test
    void clusteredSchedulersFireOnceThroughRowLock() throws Exception {
        try (LabDatabase db = new LabDatabase("org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql")) {
            QuartzFixture fixture = new QuartzFixture(db);
            Scheduler a = fixture.scheduler("u46-a", true, 7500);
            Scheduler b = fixture.scheduler("u46-b", true, 7500);
            fixture.schedule(a, "cluster-once", QuartzFixture.RecordingJob.class,
                    QuartzFixture.onceIn(700, "cluster-once-trigger"), false);
            a.start();
            b.start();
            Thread.sleep(3_000);
            assertEquals(1L, fixture.executedCount("cluster-once"), "双实例同库只允许触发一次");
            assertEquals(2L, db.count("SELECT COUNT(*) FROM QRTZ_SCHEDULER_STATE WHERE SCHED_NAME=?",
                    QuartzFixture.SCHED_NAME), "两个实例各自心跳一行");
            assertEquals(2L, db.count("SELECT COUNT(*) FROM QRTZ_LOCKS WHERE SCHED_NAME=?", QuartzFixture.SCHED_NAME),
                    "集群模式下 QRTZ_LOCKS 有 TRIGGER_ACCESS/STATE_ACCESS 两行");
            a.shutdown(false);
            b.shutdown(false);
            System.out.println("U46 quartzCluster fired=" + fixture.executedCount("cluster-once")
                    + " states=" + db.count("SELECT COUNT(*) FROM QRTZ_SCHEDULER_STATE")
                    + " locks=" + db.count("SELECT COUNT(*) FROM QRTZ_LOCKS"));
        }
    }

    @Test
    void nonClusteredModeLeavesOrphanedTriggersStuck() throws Exception {
        try (LabDatabase db = new LabDatabase("org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql")) {
            QuartzFixture fixture = new QuartzFixture(db);
            Scheduler scheduler = fixture.scheduler("u46-a", false, 7500);
            fixture.schedule(scheduler, "orphan", QuartzFixture.RecordingJob.class,
                    QuartzFixture.onceIn(60_000, "orphan-trigger"), true);
            String dead = "u46-dead";
            long stale = System.currentTimeMillis() - 20_000;
            // 重要：recoverJobs() 在调度器启动时会把所有 fired triggers 一律清空、
            // 把 ACQUIRED 一律放回 WAITING（单实例假设）。所以必须“先启动、后投放死实例行”，
            // 才能观察非集群模式缺少周期性接管。
            scheduler.start();
            Thread.sleep(500);
            db.execute("INSERT INTO QRTZ_SCHEDULER_STATE(SCHED_NAME,INSTANCE_NAME,LAST_CHECKIN_TIME,CHECKIN_INTERVAL)"
                    + " VALUES('" + QuartzFixture.SCHED_NAME + "','" + dead + "'," + stale + ",1000)");
            db.execute("INSERT INTO QRTZ_FIRED_TRIGGERS(SCHED_NAME,ENTRY_ID,TRIGGER_NAME,TRIGGER_GROUP,INSTANCE_NAME,"
                    + "FIRED_TIME,SCHED_TIME,PRIORITY,STATE,JOB_NAME,JOB_GROUP,IS_NONCONCURRENT,REQUESTS_RECOVERY)"
                    + " VALUES('" + QuartzFixture.SCHED_NAME + "','dead-entry','orphan-trigger','u46','" + dead + "',"
                    + stale + "," + stale + ",5,'ACQUIRED','orphan','u46','0','1')");
            Thread.sleep(4_000);
            assertEquals(0L, db.count("SELECT COUNT(*) FROM QRTZ_LOCKS WHERE SCHED_NAME=?",
                    QuartzFixture.SCHED_NAME), "非集群模式不建 DB 行锁行");
            assertEquals(1L, db.count("SELECT COUNT(*) FROM QRTZ_SCHEDULER_STATE WHERE INSTANCE_NAME=?", dead),
                    "非集群模式没有 ClusterManager，运行期死实例心跳行无人清理");
            assertEquals(1L, db.count("SELECT COUNT(*) FROM QRTZ_FIRED_TRIGGERS WHERE ENTRY_ID='dead-entry'"
                    + " AND STATE='ACQUIRED'"), "运行期孤立的 ACQUIRED 触发不会被周期性释放");
            scheduler.shutdown(false);
            System.out.println("U46 quartzNonCluster locks=0 deadRowKept=true orphanStillAcquired=true");
        }
    }

    @Test
    void misfireCronFireOnceNowVersusDoNothing() throws Exception {
        try (LabDatabase db = new LabDatabase("org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql")) {
            QuartzFixture fixture = new QuartzFixture(db);
            Scheduler scheduler = fixture.scheduler("u46-a", false, 7500);

            // 两个 cron 触发器都“错过”：把 NEXT_FIRE_TIME 回拨到 5 分钟前（misfireThreshold=2000ms）。
            // 用“每年 1 月 1 日 03:00”这条几乎不会在测试窗口自然触发的 cron，
            // 让“补跑次数”只由 misfire 指令决定，不受测试期间自然整点影响。
            fixture.schedule(scheduler, "cron-once", QuartzFixture.RecordingJob.class,
                    QuartzFixture.cronFireOnceNow("0 0 3 1 1 ?", "cron-once-trigger"), false);
            fixture.schedule(scheduler, "cron-nothing", QuartzFixture.RecordingJob.class,
                    QuartzFixture.cronDoNothing("0 0 3 1 1 ?", "cron-nothing-trigger"), false);
            long backdated = System.currentTimeMillis() - 300_000;
            db.execute("UPDATE QRTZ_TRIGGERS SET NEXT_FIRE_TIME=" + backdated
                    + " WHERE SCHED_NAME='" + QuartzFixture.SCHED_NAME + "'");
            scheduler.start();
            Thread.sleep(3_500);
            assertEquals(1L, fixture.executedCount("cron-once"), "FIRE_ONCE_NOW：错过多少都只补一次");
            long firedOnce = fixture.executedCount("cron-nothing");
            assertTrue(firedOnce <= 1, "DO_NOTHING 不补历史槽位（若恰好赶上新的整点秒最多 1 次）：" + firedOnce);
            long nextForNothing = nextFireTime(db, "cron-nothing-trigger");
            assertTrue(nextForNothing > System.currentTimeMillis(), "DO_NOTHING 的下次触发在未来");
            scheduler.shutdown(false);
            System.out.println("U46 misfireCron once=" + fixture.executedCount("cron-once")
                    + " nothing=" + firedOnce
                    + " nothingNextInFuture=" + (nextForNothing > System.currentTimeMillis()));
        }
    }

    @Test
    void misfireSimpleSmartPolicyReschedulesNotCatchUp() throws Exception {
        try (LabDatabase db = new LabDatabase("org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql")) {
            QuartzFixture fixture = new QuartzFixture(db);
            Scheduler scheduler = fixture.scheduler("u46-a", false, 7500);
            fixture.schedule(scheduler, "simple-smart", QuartzFixture.RecordingJob.class,
                    QuartzFixture.simpleSmart(2, "simple-smart-trigger"), false);
            long backdated = System.currentTimeMillis() - 600_000;
            db.execute("UPDATE QRTZ_TRIGGERS SET NEXT_FIRE_TIME=" + backdated
                    + " WHERE SCHED_NAME='" + QuartzFixture.SCHED_NAME + "'");
            scheduler.start();
            Thread.sleep(2_500);
            long fired = fixture.executedCount("simple-smart");
            assertTrue(fired <= 2, "SMART(repeatCount>0) 不按错过次数补跑全部 60 轮：" + fired);
            String triggerState = state(db, "simple-smart-trigger");
            long simpleRows = db.count("SELECT COUNT(*) FROM QRTZ_SIMPLE_TRIGGERS WHERE SCHED_NAME=?",
                    QuartzFixture.SCHED_NAME);
            assertTrue(simpleRows <= 1L, "simple trigger 行要么存在要么已被清理：" + simpleRows);
            scheduler.shutdown(false);
            System.out.println("U46 misfireSimple fired=" + fired + " simpleRows=" + simpleRows
                    + " state=" + triggerState);
        }
    }

    @Test
    void clusterTakesOverAfterInstanceDies() throws Exception {
        try (LabDatabase db = new LabDatabase("org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql")) {
            QuartzFixture fixture = new QuartzFixture(db);
            Scheduler survivor = fixture.scheduler("u46-survivor", true, 1000);
            fixture.schedule(survivor, "takeover", QuartzFixture.RecordingJob.class,
                    QuartzFixture.onceIn(60_000, "takeover-trigger"), true);
            survivor.start();
            Thread.sleep(500);
            // 模拟一个已经死掉的实例：它的心跳过期、并且占着一个 ACQUIRED 的触发。
            String dead = "u46-dead";
            long stale = System.currentTimeMillis() - 20_000;
            db.execute("INSERT INTO QRTZ_SCHEDULER_STATE(SCHED_NAME,INSTANCE_NAME,LAST_CHECKIN_TIME,CHECKIN_INTERVAL)"
                    + " VALUES('" + QuartzFixture.SCHED_NAME + "','" + dead + "'," + stale + ",1000)");
            db.execute("INSERT INTO QRTZ_FIRED_TRIGGERS(SCHED_NAME,ENTRY_ID,TRIGGER_NAME,TRIGGER_GROUP,INSTANCE_NAME,"
                    + "FIRED_TIME,SCHED_TIME,PRIORITY,STATE,JOB_NAME,JOB_GROUP,IS_NONCONCURRENT,REQUESTS_RECOVERY)"
                    + " VALUES('" + QuartzFixture.SCHED_NAME + "','dead-entry','orphan-trigger','u46','" + dead + "',"
                    + stale + "," + stale + ",5,'ACQUIRED','takeover','u46','0','0')");
            Thread.sleep(4_000);
            assertEquals(0L, db.count("SELECT COUNT(*) FROM QRTZ_SCHEDULER_STATE WHERE INSTANCE_NAME=?", dead),
                    "过期实例的状态行被接管者删除");
            assertEquals(0L, db.count("SELECT COUNT(*) FROM QRTZ_FIRED_TRIGGERS WHERE ENTRY_ID='dead-entry'"),
                    "孤立的 ACQUIRED 触发记录被接管者清理");
            survivor.shutdown(false);
            System.out.println("U46 clusterTakeover deadRowRemoved="
                    + (db.count("SELECT COUNT(*) FROM QRTZ_SCHEDULER_STATE WHERE INSTANCE_NAME=?", dead) == 0)
                    + " orphanCleaned=" + (db.count(
                            "SELECT COUNT(*) FROM QRTZ_FIRED_TRIGGERS WHERE ENTRY_ID='dead-entry'") == 0));
        }
    }

    @Test
    void failingJobIsNotRetriedByQuartz() throws Exception {
        try (LabDatabase db = new LabDatabase("org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql")) {
            QuartzFixture fixture = new QuartzFixture(db);
            Scheduler scheduler = fixture.scheduler("u46-a", false, 7500);
            fixture.schedule(scheduler, "failing-once", QuartzFixture.FailingJob.class,
                    QuartzFixture.onceIn(300, "failing-once-trigger"), false);
            fixture.schedule(scheduler, "failing-repeat", QuartzFixture.FailingJob.class,
                    QuartzFixture.repeatEvery(300, 2, "failing-repeat-trigger"), false);
            scheduler.start();
            Thread.sleep(2_500);
            assertEquals(1L, fixture.executedCount("failing-once"), "一次性 trigger 抛异常后不重试");
            // repeatCount=2 表示除首次外再重复 2 次，总计 3 次；若“失败即重试”会出现更多次。
            assertEquals(3L, fixture.executedCount("failing-repeat"),
                    "重复 trigger 按计划各执行一次，失败不追加额外重试");
            scheduler.shutdown(false);
            System.out.println("U46 quartzFailing noRetry once=" + fixture.executedCount("failing-once")
                    + " repeat=" + fixture.executedCount("failing-repeat"));
        }
    }

    @Test
    void disallowConcurrentBlocksRatherThanRunsParallel() throws Exception {
        try (LabDatabase db = new LabDatabase("org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql")) {
            QuartzFixture fixture = new QuartzFixture(db);
            Scheduler scheduler = fixture.scheduler("u46-a", false, 7500);
            fixture.schedule(scheduler, "nonconcurrent", QuartzFixture.SlowNonConcurrentJob.class,
                    QuartzFixture.repeatEvery(300, 3, "nonconcurrent-trigger"), false);
            scheduler.start();
            Thread.sleep(2_000);
            String triggerState = state(db, "nonconcurrent-trigger");
            assertTrue(triggerState.equals("BLOCKED") || triggerState.equals("WAITING")
                            || triggerState.equals("COMPLETE"),
                    "非并发 job 的触发器在运行期处于 BLOCKED/WAITING，而不是被并行再执行：" + triggerState);
            scheduler.shutdown(true);
            long fired = fixture.executedCount("nonconcurrent");
            assertTrue(fired >= 1 && fired <= 4, "非并发 job 不并行执行，次数不超过计划：" + fired);
            System.out.println("U46 quartzNonConcurrent fired=" + fired + " triggerState=" + triggerState);
        }
    }

    @Test
    void shutdownWaitAllowsRunningJobToFinish() throws Exception {
        try (LabDatabase db = new LabDatabase("org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql")) {
            QuartzFixture fixture = new QuartzFixture(db);
            Scheduler scheduler = fixture.scheduler("u46-a", false, 7500);
            fixture.schedule(scheduler, "slow-shutdown", QuartzFixture.SlowNonConcurrentJob.class,
                    QuartzFixture.onceIn(100, "slow-shutdown-trigger"), false);
            scheduler.start();
            Thread.sleep(600);
            assertEquals(1, fixture.running.get(), "慢 job 正在执行");
            scheduler.shutdown(true);
            assertEquals(0, fixture.running.get(), "shutdown(true) 等待在途 job 结束");
            assertEquals(1L, fixture.executedCount("slow-shutdown"));
            System.out.println("U46 quartzShutdownWait waited=true executed="
                    + fixture.executedCount("slow-shutdown"));
        }
    }

    private static String state(LabDatabase db, String trigger) throws Exception {
        return (String) db.rows("SELECT TRIGGER_STATE FROM QRTZ_TRIGGERS WHERE SCHED_NAME=? AND TRIGGER_NAME=?",
                QuartzFixture.SCHED_NAME, trigger).get(0).get("TRIGGER_STATE");
    }

    private static long nextFireTime(LabDatabase db, String trigger) throws Exception {
        Object value = db.rows("SELECT NEXT_FIRE_TIME FROM QRTZ_TRIGGERS WHERE SCHED_NAME=? AND TRIGGER_NAME=?",
                QuartzFixture.SCHED_NAME, trigger).get(0).get("NEXT_FIRE_TIME");
        return value == null ? Long.MAX_VALUE : ((Number) value).longValue();
    }
}