package com.easyagents.scheduler.quartz;

import com.easyagents.scheduler.CronSchedulePlan;
import com.easyagents.scheduler.OnceSchedulePlan;
import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleFireReceipt;
import com.easyagents.scheduler.ScheduleId;
import com.easyagents.scheduler.SchedulePlan;
import com.easyagents.scheduler.ScheduleService;
import com.easyagents.scheduler.ScheduleStatus;
import com.easyagents.scheduler.ScheduleView;
import org.quartz.CronExpression;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.ObjectAlreadyExistsException;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.Trigger.TriggerState;
import org.quartz.TriggerKey;

import java.text.ParseException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 基于单个 Quartz Scheduler 的调度服务实现。
 */
public final class QuartzScheduleService implements ScheduleService, AutoCloseable {

    /** 立即触发调用标识的最大长度。 */
    public static final int MAX_INVOCATION_ID_LENGTH = 190;

    /** 单次触发时间预览的最大数量。 */
    public static final int MAX_PREVIEW_LIMIT = 100;

    private final Scheduler scheduler;
    private final boolean waitForJobsToCompleteOnShutdown;
    private final long shutdownWaitTimeoutMillis;
    private final AtomicBoolean closed = new AtomicBoolean();

    /**
     * 创建 Quartz 调度服务。
     *
     * @param scheduler 已完成运行时装配、尚未启动的 Scheduler
     * @param waitForJobsToCompleteOnShutdown 关闭时是否等待在途任务
     * @param shutdownWaitTimeoutMillis 关闭等待最长毫秒数
     */
    QuartzScheduleService(
        Scheduler scheduler,
        boolean waitForJobsToCompleteOnShutdown,
        long shutdownWaitTimeoutMillis
    ) {
        this.scheduler = Objects.requireNonNull(scheduler, "scheduler must not be null");
        this.waitForJobsToCompleteOnShutdown = waitForJobsToCompleteOnShutdown;
        if (shutdownWaitTimeoutMillis < 1) {
            throw new IllegalArgumentException("shutdownWaitTimeoutMillis must be positive");
        }
        this.shutdownWaitTimeoutMillis = shutdownWaitTimeoutMillis;
    }

    /**
     * 启动底层 Quartz Scheduler。
     *
     * @throws ScheduleException Scheduler 启动失败或已经关闭时抛出
     */
    public void start() {
        ensureOpen();
        try {
            if (!scheduler.isStarted()) {
                scheduler.start();
            }
        } catch (SchedulerException exception) {
            ScheduleException failure = providerFailure(
                "failed to start Quartz scheduler",
                exception
            );
            closed.set(true);
            try {
                if (!scheduler.isShutdown()) {
                    scheduler.shutdown(false);
                }
            } catch (SchedulerException cleanupFailure) {
                failure.addSuppressed(cleanupFailure);
            }
            throw failure;
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public ScheduleView create(ScheduleDefinition definition) {
        Objects.requireNonNull(definition, "definition must not be null");
        ensureOpen();
        JobDetail job = mapJob(definition);
        Trigger trigger = mapTrigger(definition);
        try {
            scheduler.scheduleJob(job, trigger);
            return requireView(definition.id());
        } catch (ObjectAlreadyExistsException exception) {
            Optional<ScheduleView> existing = get(definition.id());
            if (existing.isPresent() && existing.get().definition().equals(definition)) {
                return existing.get();
            }
            throw new ScheduleException(
                ScheduleErrorCode.SCHEDULE_CONFLICT,
                "schedule already exists with a different definition: " + definition.id(),
                exception
            );
        } catch (SchedulerException exception) {
            throw providerFailure("failed to create schedule " + definition.id(), exception);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public ScheduleView replace(ScheduleDefinition definition) {
        Objects.requireNonNull(definition, "definition must not be null");
        ensureOpen();
        JobDetail job = mapJob(definition);
        Trigger trigger = mapTrigger(definition);
        try {
            // Quartz 在一个 JobStore 事务中替换 Job 与其唯一 Trigger，避免中间态。
            scheduler.scheduleJob(job, Set.of(trigger), true);
            return requireView(definition.id());
        } catch (SchedulerException exception) {
            throw providerFailure("failed to replace schedule " + definition.id(), exception);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public ScheduleView pause(ScheduleId scheduleId) {
        Objects.requireNonNull(scheduleId, "scheduleId must not be null");
        ensureOpen();
        try {
            if (!scheduler.checkExists(QuartzScheduleMapper.jobKey(scheduleId))) {
                throw notFound(scheduleId);
            }
            TriggerKey triggerKey = QuartzScheduleMapper.triggerKey(scheduleId);
            if (scheduler.checkExists(triggerKey)) {
                scheduler.pauseTrigger(triggerKey);
            }
            return requireView(scheduleId);
        } catch (ScheduleException exception) {
            throw exception;
        } catch (SchedulerException exception) {
            throw providerFailure("failed to pause schedule " + scheduleId, exception);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public ScheduleView resume(ScheduleId scheduleId) {
        Objects.requireNonNull(scheduleId, "scheduleId must not be null");
        ensureOpen();
        try {
            if (!scheduler.checkExists(QuartzScheduleMapper.jobKey(scheduleId))) {
                throw notFound(scheduleId);
            }
            TriggerKey triggerKey = QuartzScheduleMapper.triggerKey(scheduleId);
            if (scheduler.checkExists(triggerKey)) {
                scheduler.resumeTrigger(triggerKey);
            }
            return requireView(scheduleId);
        } catch (ScheduleException exception) {
            throw exception;
        } catch (SchedulerException exception) {
            throw providerFailure("failed to resume schedule " + scheduleId, exception);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean delete(ScheduleId scheduleId) {
        Objects.requireNonNull(scheduleId, "scheduleId must not be null");
        ensureOpen();
        try {
            // 批量接口对单个 Job 也直接委派 JobStore，避免逐 Trigger 删除时的完成竞态。
            return scheduler.deleteJobs(List.of(QuartzScheduleMapper.jobKey(scheduleId)));
        } catch (SchedulerException exception) {
            throw providerFailure("failed to delete schedule " + scheduleId, exception);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public ScheduleFireReceipt triggerNow(
        ScheduleId scheduleId,
        String invocationId,
        Map<String, String> parameters
    ) {
        Objects.requireNonNull(scheduleId, "scheduleId must not be null");
        String normalizedInvocationId = normalizeInvocationId(invocationId);
        ensureOpen();
        try {
            JobDetail job = scheduler.getJobDetail(QuartzScheduleMapper.jobKey(scheduleId));
            if (job == null) {
                throw notFound(scheduleId);
            }
            Map<String, String> mergedParameters = new HashMap<>(
                QuartzScheduleMapper.toDefinition(job.getJobDataMap()).parameters()
            );
            if (parameters != null) {
                mergedParameters.putAll(parameters);
            }
            Map<String, String> safeParameters = ScheduleDefinition
                .copyAndValidateParameters(mergedParameters);
            JobDataMap data = QuartzScheduleMapper.immediateData(
                scheduleId,
                normalizedInvocationId,
                safeParameters
            );
            scheduler.triggerJob(QuartzScheduleMapper.jobKey(scheduleId), data);
            return new ScheduleFireReceipt(scheduleId, normalizedInvocationId, Instant.now());
        } catch (ScheduleException exception) {
            throw exception;
        } catch (SchedulerException exception) {
            throw providerFailure("failed to trigger schedule " + scheduleId, exception);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Optional<ScheduleView> get(ScheduleId scheduleId) {
        Objects.requireNonNull(scheduleId, "scheduleId must not be null");
        ensureOpen();
        try {
            JobDetail job = scheduler.getJobDetail(QuartzScheduleMapper.jobKey(scheduleId));
            if (job == null) {
                return Optional.empty();
            }
            ScheduleDefinition definition = QuartzScheduleMapper.toDefinition(job.getJobDataMap());
            TriggerKey triggerKey = QuartzScheduleMapper.triggerKey(scheduleId);
            Trigger trigger = scheduler.getTrigger(triggerKey);
            TriggerState triggerState = trigger == null
                ? TriggerState.NONE
                : scheduler.getTriggerState(triggerKey);
            return Optional.of(new ScheduleView(
                definition,
                toStatus(triggerState, trigger == null),
                toInstant(trigger == null ? null : trigger.getPreviousFireTime()),
                toInstant(trigger == null ? null : trigger.getNextFireTime())
            ));
        } catch (SchedulerException exception) {
            throw providerFailure("failed to read schedule " + scheduleId, exception);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public List<Instant> nextFireTimes(SchedulePlan plan, int limit) {
        Objects.requireNonNull(plan, "plan must not be null");
        if (limit < 1 || limit > MAX_PREVIEW_LIMIT) {
            throw new ScheduleException(
                ScheduleErrorCode.INVALID_DEFINITION,
                "limit must be between 1 and " + MAX_PREVIEW_LIMIT
            );
        }
        if (plan instanceof OnceSchedulePlan oncePlan) {
            return oncePlan.fireAt().isAfter(Instant.now())
                ? List.of(oncePlan.fireAt())
                : List.of();
        }
        if (plan instanceof CronSchedulePlan cronPlan) {
            return nextCronFireTimes(cronPlan, limit);
        }
        throw new ScheduleException(
            ScheduleErrorCode.INVALID_DEFINITION,
            "unsupported schedule plan: " + plan.getClass().getName()
        );
    }

    /**
     * 关闭 Scheduler 并释放 Provider 对外部 DataSource 的强引用。
     *
     * <p>外部 DataSource 的连接池所有权仍属于调用方，本方法不会关闭连接池。</p>
     *
     * @throws ScheduleException Scheduler 关闭失败时抛出
     */
    @Override
    public void close() {
        if (!closed.compareAndSet(false, true)) {
            return;
        }
        ScheduleException failure = null;
        try {
            if (!scheduler.isShutdown()) {
                scheduler.standby();
                if (
                    waitForJobsToCompleteOnShutdown
                        && !awaitExecutingJobs(shutdownWaitTimeoutMillis)
                ) {
                    failure = new ScheduleException(
                        ScheduleErrorCode.PROVIDER_FAILURE,
                        "timed out after " + shutdownWaitTimeoutMillis
                            + " ms waiting for Quartz jobs to complete"
                    );
                }
            }
        } catch (InterruptedException exception) {
            Thread.currentThread().interrupt();
            failure = providerFailure(
                "interrupted while waiting for Quartz jobs to complete",
                exception
            );
        } catch (SchedulerException exception) {
            failure = providerFailure("failed to prepare Quartz scheduler shutdown", exception);
        }
        try {
            if (!scheduler.isShutdown()) {
                // Factory 启用 interruptJobsOnShutdown，Quartz 会先中断内部
                // InterruptableJob；false 只表示不再无界等待忽略中断的 Handler。
                scheduler.shutdown(false);
            }
        } catch (SchedulerException exception) {
            if (failure == null) {
                failure = providerFailure("failed to close Quartz scheduler", exception);
            } else {
                failure.addSuppressed(exception);
            }
        }
        if (failure != null) {
            throw failure;
        }
    }

    /**
     * 返回底层 Scheduler，供 Provider 诊断和测试使用。
     *
     * @return 当前 Quartz Scheduler
     */
    Scheduler quartzScheduler() {
        return scheduler;
    }

    /**
     * 在给定时限内等待当前节点任务完成。
     *
     * @param timeoutMillis 最长等待毫秒数
     * @return 全部完成时为 true，超时时为 false
     * @throws SchedulerException 读取执行状态失败时抛出
     * @throws InterruptedException 等待线程被中断时抛出
     */
    private boolean awaitExecutingJobs(long timeoutMillis)
        throws SchedulerException, InterruptedException {
        long deadline = System.nanoTime() + java.util.concurrent.TimeUnit.MILLISECONDS.toNanos(
            timeoutMillis
        );
        while (!scheduler.getCurrentlyExecutingJobs().isEmpty()) {
            long remainingNanos = deadline - System.nanoTime();
            if (remainingNanos <= 0L) {
                return false;
            }
            long sleepMillis = Math.max(
                1L,
                Math.min(
                    100L,
                    java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(remainingNanos)
                )
            );
            Thread.sleep(sleepMillis);
        }
        return true;
    }

    /**
     * 查询必然存在的任务视图。
     *
     * @param scheduleId 调度标识
     * @return 任务视图
     */
    private ScheduleView requireView(ScheduleId scheduleId) {
        return get(scheduleId).orElseThrow(() -> notFound(scheduleId));
    }

    /**
     * 映射 Job 并统一定义错误。
     *
     * @param definition 调度定义
     * @return Job 详情
     */
    private static JobDetail mapJob(ScheduleDefinition definition) {
        try {
            return QuartzScheduleMapper.toJobDetail(definition);
        } catch (ScheduleException exception) {
            throw exception;
        } catch (RuntimeException exception) {
            throw invalidDefinition(definition.id(), exception);
        }
    }

    /**
     * 映射 Trigger 并统一定义错误。
     *
     * @param definition 调度定义
     * @return Quartz Trigger
     */
    private static Trigger mapTrigger(ScheduleDefinition definition) {
        try {
            return QuartzScheduleMapper.toTrigger(definition);
        } catch (ScheduleException exception) {
            throw exception;
        } catch (RuntimeException exception) {
            throw invalidDefinition(definition.id(), exception);
        }
    }

    /**
     * 计算 Cron 后续触发时间。
     *
     * @param plan Cron 计划
     * @param limit 最大数量
     * @return 触发时间
     */
    private static List<Instant> nextCronFireTimes(CronSchedulePlan plan, int limit) {
        try {
            CronExpression expression = new CronExpression(plan.expression());
            expression.setTimeZone(java.util.TimeZone.getTimeZone(plan.zoneId()));
            List<Instant> fireTimes = new ArrayList<>(limit);
            Date cursor = Date.from(Instant.now());
            for (int index = 0; index < limit; index++) {
                Date next = expression.getNextValidTimeAfter(cursor);
                if (next == null) {
                    break;
                }
                fireTimes.add(next.toInstant());
                cursor = next;
            }
            return List.copyOf(fireTimes);
        } catch (ParseException exception) {
            throw new ScheduleException(
                ScheduleErrorCode.INVALID_DEFINITION,
                "invalid Quartz cron expression: " + plan.expression(),
                exception
            );
        }
    }

    /**
     * 映射 Quartz Trigger 状态。
     *
     * @param state Quartz 状态
     * @param completed 是否已无主 Trigger
     * @return 公共状态
     */
    private static ScheduleStatus toStatus(TriggerState state, boolean completed) {
        if (completed) {
            return ScheduleStatus.COMPLETE;
        }
        return switch (state) {
            case NORMAL -> ScheduleStatus.SCHEDULED;
            case PAUSED -> ScheduleStatus.PAUSED;
            case BLOCKED -> ScheduleStatus.BLOCKED;
            case ERROR -> ScheduleStatus.ERROR;
            case COMPLETE, NONE -> ScheduleStatus.COMPLETE;
        };
    }

    /**
     * 将 JDBC/Quartz 时间转换为 Instant。
     *
     * @param value 日期，可为空
     * @return Instant，可为空
     */
    private static Instant toInstant(Date value) {
        return value == null ? null : value.toInstant();
    }

    /**
     * 规范化立即触发调用标识。
     *
     * @param invocationId 原始调用标识
     * @return 规范化标识
     */
    private static String normalizeInvocationId(String invocationId) {
        if (invocationId == null || invocationId.isBlank()) {
            throw new ScheduleException(
                ScheduleErrorCode.INVALID_DEFINITION,
                "invocationId must not be blank"
            );
        }
        String normalized = invocationId.trim();
        if (normalized.length() > MAX_INVOCATION_ID_LENGTH) {
            throw new ScheduleException(
                ScheduleErrorCode.INVALID_DEFINITION,
                "invocationId length must not exceed " + MAX_INVOCATION_ID_LENGTH
            );
        }
        return normalized;
    }

    /**
     * 确认服务仍可接受操作。
     */
    private void ensureOpen() {
        if (closed.get()) {
            throw new ScheduleException(
                ScheduleErrorCode.SCHEDULER_CLOSED,
                "scheduler is already closed"
            );
        }
    }

    /**
     * 构造任务不存在异常。
     *
     * @param scheduleId 调度标识
     * @return 稳定异常
     */
    private static ScheduleException notFound(ScheduleId scheduleId) {
        return new ScheduleException(
            ScheduleErrorCode.SCHEDULE_NOT_FOUND,
            "schedule not found: " + scheduleId
        );
    }

    /**
     * 构造定义非法异常。
     *
     * @param scheduleId 调度标识
     * @param exception 原始原因
     * @return 稳定异常
     */
    private static ScheduleException invalidDefinition(
        ScheduleId scheduleId,
        RuntimeException exception
    ) {
        return new ScheduleException(
            ScheduleErrorCode.INVALID_DEFINITION,
            "invalid schedule definition: " + scheduleId,
            exception
        );
    }

    /**
     * 构造 Provider 失败异常。
     *
     * @param message 失败摘要
     * @param exception 原始原因
     * @return 稳定异常
     */
    private static ScheduleException providerFailure(String message, Exception exception) {
        return new ScheduleException(ScheduleErrorCode.PROVIDER_FAILURE, message, exception);
    }
}
