package study.batchquartz;

import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobBuilder;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.impl.SchedulerRepository;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.utils.DBConnectionManager;

/// U46 的 Quartz 基座：真实 StdSchedulerFactory + JDBC JobStore（JobStoreTX，官方
/// tables_mysql_innodb.sql 已由 LabDatabase 执行）；同一库、同一 SCHED_NAME 可起多个
/// Scheduler 对象模拟同机多实例，行锁走真实 MySQL。
///
/// 两个真实坑：
/// 1) `SchedulerRepository` 按 scheduler 名全局去重；同 JVM 第二实例必须先移除仓库条目，
///    否则直接抛“Scheduler with name already exists”。移除只影响查找，不影响已建实例。
/// 2) `DBConnectionManager` 按 dataSource 名全局缓存 provider；每个实例用独立 dataSource 名，
///    避免第二个实例复用第一个的连接池。
final class QuartzFixture {

    static final String SCHED_NAME = "u46-lab";
    private static final ConcurrentHashMap<String, QuartzFixture> ACTIVE = new ConcurrentHashMap<>();

    final LabDatabase db;
    final CopyOnWriteArrayList<String> executed = new CopyOnWriteArrayList<>();
    final AtomicInteger concurrentPeak = new AtomicInteger();
    final AtomicInteger running = new AtomicInteger();

    QuartzFixture(LabDatabase db) {
        this.db = db;
    }

    Scheduler scheduler(String instanceId, boolean clustered, long checkinInterval) throws SchedulerException {
        String dsName = "u46-" + instanceId;
        Properties props = new Properties();
        props.setProperty("org.quartz.scheduler.instanceName", SCHED_NAME);
        props.setProperty("org.quartz.scheduler.instanceId", instanceId);
        props.setProperty("org.quartz.scheduler.skipUpdateCheck", "true");
        props.setProperty("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool");
        props.setProperty("org.quartz.threadPool.threadCount", "4");
        props.setProperty("org.quartz.jobStore.class", "org.quartz.impl.jdbcjobstore.JobStoreTX");
        props.setProperty("org.quartz.jobStore.driverDelegateClass",
                "org.quartz.impl.jdbcjobstore.StdJDBCDelegate");
        props.setProperty("org.quartz.jobStore.tablePrefix", "QRTZ_");
        props.setProperty("org.quartz.jobStore.dataSource", dsName);
        props.setProperty("org.quartz.jobStore.isClustered", Boolean.toString(clustered));
        props.setProperty("org.quartz.jobStore.clusterCheckinInterval", Long.toString(checkinInterval));
        props.setProperty("org.quartz.jobStore.misfireThreshold", "2000");
        props.setProperty("org.quartz.dataSource." + dsName + ".provider", "hikaricp");
        props.setProperty("org.quartz.dataSource." + dsName + ".driver", "com.mysql.cj.jdbc.Driver");
        props.setProperty("org.quartz.dataSource." + dsName + ".URL", this.db.url);
        props.setProperty("org.quartz.dataSource." + dsName + ".user", LabDatabase.USER);
        props.setProperty("org.quartz.dataSource." + dsName + ".password", LabDatabase.PASSWORD);
        props.setProperty("org.quartz.dataSource." + dsName + ".maxConnections", "5");
        SchedulerRepository.getInstance().remove(SCHED_NAME);
        StdSchedulerFactory factory = new StdSchedulerFactory();
        factory.initialize(props);
        return factory.getScheduler();
    }

    JobKey schedule(Scheduler scheduler, String name, Class<? extends Job> jobClass,
            Trigger trigger, boolean requestsRecovery) throws SchedulerException {
        String ref = SCHED_NAME + "/" + this.db.name + "/" + scheduler.getSchedulerInstanceId();
        ACTIVE.put(ref, this);
        JobDataMap data = new JobDataMap();
        data.put("fixtureRef", ref);
        var detail = JobBuilder.newJob(jobClass)
                .withIdentity(name, "u46")
                .usingJobData(data)
                .requestRecovery(requestsRecovery)
                .build();
        scheduler.scheduleJob(detail, trigger);
        return detail.getKey();
    }

    private static QuartzFixture active(JobExecutionContext context) {
        return ACTIVE.get(context.getMergedJobDataMap().getString("fixtureRef"));
    }

    static Trigger onceIn(long millis, String name) {
        return TriggerBuilder.newTrigger()
                .withIdentity(name, "u46")
                .startAt(new java.util.Date(System.currentTimeMillis() + millis))
                .build();
    }

    static Trigger repeatEvery(long millis, int repeatCount, String name) {
        return TriggerBuilder.newTrigger()
                .withIdentity(name, "u46")
                .startAt(new java.util.Date(System.currentTimeMillis() + 200))
                .withSchedule(SimpleScheduleBuilder.simpleSchedule()
                        .withIntervalInMilliseconds(millis)
                        .withRepeatCount(repeatCount)
                        .withMisfireHandlingInstructionNextWithRemainingCount())
                .build();
    }

    /// cron + MISFIRE_INSTRUCTION_FIRE_ONCE_NOW（缺省 SMART 对 CronTrigger 就是它，这里显式写清）。
    static Trigger cronFireOnceNow(String expression, String name) {
        return TriggerBuilder.newTrigger()
                .withIdentity(name, "u46")
                .withSchedule(org.quartz.CronScheduleBuilder.cronSchedule(expression)
                        .withMisfireHandlingInstructionFireAndProceed())
                .build();
    }

    /// cron + MISFIRE_INSTRUCTION_DO_NOTHING。
    static Trigger cronDoNothing(String expression, String name) {
        return TriggerBuilder.newTrigger()
                .withIdentity(name, "u46")
                .withSchedule(org.quartz.CronScheduleBuilder.cronSchedule(expression)
                        .withMisfireHandlingInstructionDoNothing())
                .build();
    }

    /// SimpleTrigger + SMART(repeatCount>0) 的等价指令 RESCHEDULE_NOW_WITH_EXISTING_COUNT；
    /// NEXT_FIRE_TIME 由测试回拨到 2 分钟前，让触发器处于“已错过”。
    static Trigger simpleSmart(int repeatCount, String name) {
        return TriggerBuilder.newTrigger()
                .withIdentity(name, "u46")
                .withSchedule(SimpleScheduleBuilder.simpleSchedule()
                        .withIntervalInSeconds(10)
                        .withRepeatCount(repeatCount)
                        .withMisfireHandlingInstructionNowWithExistingCount())
                .startAt(new java.util.Date(System.currentTimeMillis() + 1_000))
                .build();
    }

    long executedCount(String name) {
        return this.executed.stream().filter(line -> line.startsWith(name + "@")).count();
    }

    /// 记录型 job：写在内存，不产生外部副作用。
    public static class RecordingJob implements Job {
        @Override
        public void execute(JobExecutionContext context) throws JobExecutionException {
            QuartzFixture fixture = active(context);
            int now = fixture.running.incrementAndGet();
            fixture.concurrentPeak.accumulateAndGet(now, Math::max);
            try {
                fixture.executed.add(context.getJobDetail().getKey().getName() + "@"
                        + context.getScheduler().getSchedulerInstanceId() + "#"
                        + context.getRefireCount());
                Thread.sleep(30);
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            catch (SchedulerException e) {
                throw new JobExecutionException(e);
            }
            finally {
                fixture.running.decrementAndGet();
            }
        }
    }

    /// 永不并发的 job：验证 @DisallowConcurrentExecution 的 trigger BLOCKED 语义。
    @DisallowConcurrentExecution
    public static class SlowNonConcurrentJob implements Job {
        @Override
        public void execute(JobExecutionContext context) throws JobExecutionException {
            QuartzFixture fixture = active(context);
            fixture.running.incrementAndGet();
            try {
                fixture.executed.add(context.getJobDetail().getKey().getName() + "@"
                        + context.getScheduler().getSchedulerInstanceId());
                Thread.sleep(1200);
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            catch (SchedulerException e) {
                throw new JobExecutionException(e);
            }
            finally {
                fixture.running.decrementAndGet();
            }
        }
    }

    /// 抛异常的 job：验证 Quartz 不会自动重试。
    public static class FailingJob implements Job {
        @Override
        public void execute(JobExecutionContext context) throws JobExecutionException {
            QuartzFixture fixture = active(context);
            String instance;
            try {
                instance = context.getScheduler().getSchedulerInstanceId();
            }
            catch (SchedulerException e) {
                instance = "unknown";
            }
            fixture.executed.add(context.getJobDetail().getKey().getName() + "@" + instance + "#fail");
            throw new JobExecutionException(new IllegalStateException("synthetic quartz failure"), false);
        }
    }
}