package com.easyagents.scheduler.quartz;

import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleExecutionListener;
import com.easyagents.scheduler.ScheduleFireContext;
import com.easyagents.scheduler.ScheduleHandler;
import com.easyagents.scheduler.ScheduleId;
import com.easyagents.scheduler.ScheduleRefireException;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.ObjectAlreadyExistsException;
import org.quartz.SchedulerException;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.TriggerKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.time.Instant;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.locks.LockSupport;

/**
 * 当前 Scheduler 节点的 Handler 和监听器运行时。
 */
final class QuartzRuntime {

    static final String SCHEDULER_CONTEXT_KEY = "easyAgentsSchedulerRuntime";

    private static final Logger log = LoggerFactory.getLogger(QuartzRuntime.class);

    private final Map<String, ScheduleHandler> handlers;
    private final List<ScheduleExecutionListener> listeners;

    /**
     * 创建节点本地运行时。
     *
     * @param handlers 当前节点 Handler
     * @param listeners 当前节点执行监听器
     */
    QuartzRuntime(
        Collection<? extends ScheduleHandler> handlers,
        Collection<? extends ScheduleExecutionListener> listeners
    ) {
        this.handlers = registerHandlers(handlers);
        this.listeners = listeners == null ? List.of() : List.copyOf(listeners);
    }

    /**
     * 执行 Quartz 触发并保留 Handler 原始失败原因。
     *
     * @param quartzContext Quartz 执行上下文
     * @throws JobExecutionException Handler 或运行时执行失败时抛出
     */
    void execute(JobExecutionContext quartzContext) throws JobExecutionException {
        JobDataMap data = quartzContext.getMergedJobDataMap();
        ScheduleDefinition definition = QuartzScheduleMapper.toDefinition(data);
        Instant observedActualFireTime = toInstant(quartzContext.getFireTime(), Instant.now());
        Instant scheduledFireTime = instantValue(
            data,
            QuartzScheduleMapper.KEY_RETRY_SCHEDULED_FIRE_TIME,
            toInstant(quartzContext.getScheduledFireTime(), observedActualFireTime)
        );
        Instant actualFireTime = instantValue(
            data,
            QuartzScheduleMapper.KEY_RETRY_ACTUAL_FIRE_TIME,
            observedActualFireTime
        );
        ScheduleFireContext context = new ScheduleFireContext(
            definition.id(),
            definition.handlerCode(),
            scheduledFireTime,
            actualFireTime,
            stringValue(data, QuartzScheduleMapper.KEY_RETRY_FIRE_INSTANCE_ID,
                quartzContext.getFireInstanceId()),
            stringValue(data, QuartzScheduleMapper.KEY_INVOCATION),
            booleanValue(data, QuartzScheduleMapper.KEY_RETRY_RECOVERING,
                quartzContext.isRecovering()),
            definition.parameters()
        );
        ScheduleHandler handler = handlers.get(definition.handlerCode());
        if (handler == null) {
            ScheduleException failure = new ScheduleException(
                ScheduleErrorCode.HANDLER_NOT_FOUND,
                "schedule handler not found: " + definition.handlerCode()
            );
            notifyFailed(context, Duration.ZERO, failure);
            throw new JobExecutionException(failure, false);
        }
        notifyStarted(context);
        long startedAt = System.nanoTime();
        try {
            handler.execute(context);
            notifySucceeded(context, elapsed(startedAt));
        } catch (Exception exception) {
            notifyFailed(context, elapsed(startedAt), exception);
            if (exception instanceof ScheduleRefireException retry) {
                int attempt = Math.max(
                    intValue(data, QuartzScheduleMapper.KEY_RETRY_ATTEMPT, 0),
                    quartzContext.getRefireCount()
                );
                if (retry.maxRefires() == 0 || attempt < retry.maxRefires()) {
                    try {
                        persistRetry(quartzContext, context, retry, attempt + 1);
                        return;
                    } catch (SchedulerException retryFailure) {
                        exception.addSuppressed(retryFailure);
                        log.error("Failed to persist schedule handler retry: scheduleId={}, attempt={}",
                            context.scheduleId(), attempt + 1, retryFailure);
                        if (canRefireInPlace(quartzContext, retry, attempt + 1)) {
                            // JobStore 暂时不可写时，正常完成原 fire 会造成登记丢失。
                            // 仅在这条降级路径短时占用当前 worker；一旦持久重试落库即释放。
                            throw new JobExecutionException(exception, true);
                        }
                    }
                } else {
                    log.error("Schedule handler retry limit exhausted: scheduleId={}, retries={}",
                        context.scheduleId(), retry.maxRefires(), exception);
                }
            }
            // 不在 Quartz worker 内无限 refire。持久化重试失败时保留原异常，
            // 由 requestRecovery 和集群故障恢复处理未完成的 fired trigger。
            throw new JobExecutionException(exception, false);
        }
    }

    private static boolean canRefireInPlace(
        JobExecutionContext context,
        ScheduleRefireException retry,
        int attempt
    ) {
        LockSupport.parkNanos(retry.delayFor(attempt).toNanos());
        if (Thread.currentThread().isInterrupted()) return false;
        try {
            return !context.getScheduler().isShutdown();
        } catch (SchedulerException exception) {
            return false;
        }
    }

    private static void persistRetry(
        JobExecutionContext quartzContext,
        ScheduleFireContext context,
        ScheduleRefireException retry,
        int attempt
    ) throws SchedulerException {
        String source = context.scheduleId() + "|" + context.fireInstanceId() + "|"
            + String.valueOf(context.invocationId()) + "|" + attempt;
        String retryName = "retry-" + UUID.nameUUIDFromBytes(
            source.getBytes(StandardCharsets.UTF_8));
        TriggerKey retryKey = new TriggerKey(
            retryName,
            QuartzScheduleMapper.GROUP_PREFIX + "retry." + context.scheduleId().namespace()
        );
        Trigger trigger = TriggerBuilder.newTrigger()
            .withIdentity(retryKey)
            .forJob(quartzContext.getJobDetail().getKey())
            .usingJobData(QuartzScheduleMapper.retryData(context, attempt))
            .startAt(Date.from(Instant.now().plus(retry.delayFor(attempt))))
            .withSchedule(SimpleScheduleBuilder.simpleSchedule()
                .withRepeatCount(0)
                .withMisfireHandlingInstructionFireNow())
            .build();
        try {
            quartzContext.getScheduler().scheduleJob(trigger);
        } catch (ObjectAlreadyExistsException ignored) {
            // 同一原始 fire/attempt 的确定性 key 已落库，即视为持久化成功。
        }
    }

    /**
     * 向监听器发布 Misfire 事件。
     *
     * @param scheduleId 任务标识
     * @param expectedFireTime 原计划时间，可为空
     */
    void notifyMisfired(ScheduleId scheduleId, Instant expectedFireTime) {
        for (ScheduleExecutionListener listener : listeners) {
            try {
                listener.onMisfired(scheduleId, expectedFireTime);
            } catch (RuntimeException exception) {
                log.warn("Schedule listener failed on misfire: {}", scheduleId, exception);
            }
        }
    }

    private void notifyStarted(ScheduleFireContext context) {
        for (ScheduleExecutionListener listener : listeners) {
            try {
                listener.onStarted(context);
            } catch (RuntimeException exception) {
                log.warn("Schedule listener failed on start: {}", context.scheduleId(), exception);
            }
        }
    }

    private void notifySucceeded(ScheduleFireContext context, Duration duration) {
        for (ScheduleExecutionListener listener : listeners) {
            try {
                listener.onSucceeded(context, duration);
            } catch (RuntimeException exception) {
                log.warn("Schedule listener failed on success: {}", context.scheduleId(), exception);
            }
        }
    }

    private void notifyFailed(
        ScheduleFireContext context,
        Duration duration,
        Throwable failure
    ) {
        for (ScheduleExecutionListener listener : listeners) {
            try {
                listener.onFailed(context, duration, failure);
            } catch (RuntimeException exception) {
                log.warn("Schedule listener failed on failure: {}", context.scheduleId(), exception);
            }
        }
    }

    private static Map<String, ScheduleHandler> registerHandlers(
        Collection<? extends ScheduleHandler> handlers
    ) {
        Map<String, ScheduleHandler> registered = new LinkedHashMap<>();
        if (handlers == null) {
            return Map.of();
        }
        for (ScheduleHandler handler : handlers) {
            String rawCode = handler == null ? null : handler.code();
            if (rawCode == null || rawCode.isBlank()) {
                throw new ScheduleException(
                    ScheduleErrorCode.INVALID_DEFINITION,
                    "schedule handler and handler code must not be blank"
                );
            }
            String code = rawCode.trim();
            ScheduleHandler previous = registered.putIfAbsent(code, handler);
            if (previous != null) {
                throw new ScheduleException(
                    ScheduleErrorCode.DUPLICATE_HANDLER,
                    "duplicate schedule handler code: " + code
                );
            }
        }
        return Map.copyOf(registered);
    }

    private static Duration elapsed(long startedAt) {
        return Duration.ofNanos(Math.max(0L, System.nanoTime() - startedAt));
    }

    private static Instant toInstant(java.util.Date value, Instant fallback) {
        return value == null ? fallback : value.toInstant();
    }

    private static String stringValue(JobDataMap data, String key) {
        Object value = data.get(key);
        return value == null ? null : value.toString();
    }

    private static String stringValue(JobDataMap data, String key, String fallback) {
        String value = stringValue(data, key);
        return value == null ? fallback : value;
    }

    private static Instant instantValue(JobDataMap data, String key, Instant fallback) {
        String value = stringValue(data, key);
        return value == null ? fallback : Instant.ofEpochMilli(Long.parseLong(value));
    }

    private static boolean booleanValue(JobDataMap data, String key, boolean fallback) {
        String value = stringValue(data, key);
        return value == null ? fallback : Boolean.parseBoolean(value);
    }

    private static int intValue(JobDataMap data, String key, int fallback) {
        String value = stringValue(data, key);
        return value == null ? fallback : Integer.parseInt(value);
    }
}
