package com.easyagents.scheduler.quartz;

import com.easyagents.scheduler.ConcurrencyPolicy;
import com.easyagents.scheduler.CronSchedulePlan;
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.ScheduleId;
import com.easyagents.scheduler.SchedulePlan;
import org.quartz.CronScheduleBuilder;
import org.quartz.JobBuilder;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.TriggerKey;

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Base64;
import java.util.HashMap;
import java.util.Map;
import java.util.TimeZone;

/**
 * Core 调度模型与 Quartz 持久模型之间的单一映射器。
 */
final class QuartzScheduleMapper {

    static final String SCHEMA_VERSION = "1";
    static final String KEY_SCHEMA_VERSION = "ea.schemaVersion";
    static final String KEY_NAMESPACE = "ea.namespace";
    static final String KEY_NAME = "ea.name";
    static final String KEY_HANDLER = "ea.handler";
    static final String KEY_PLAN_TYPE = "ea.planType";
    static final String KEY_CRON = "ea.cron";
    static final String KEY_ZONE = "ea.zone";
    static final String KEY_FIRE_AT = "ea.fireAt";
    static final String KEY_MISFIRE = "ea.misfire";
    static final String KEY_CONCURRENCY = "ea.concurrency";
    static final String KEY_RECOVER = "ea.recover";
    static final String KEY_DESCRIPTION = "ea.description";
    static final String KEY_INVOCATION = "ea.invocationId";
    static final String KEY_IMMEDIATE_PARAMETER_SNAPSHOT = "ea.immediateParameterSnapshot";
    static final String KEY_RETRY_ATTEMPT = "ea.retry.attempt";
    static final String KEY_RETRY_SCHEDULED_FIRE_TIME = "ea.retry.scheduledFireTime";
    static final String KEY_RETRY_ACTUAL_FIRE_TIME = "ea.retry.actualFireTime";
    static final String KEY_RETRY_FIRE_INSTANCE_ID = "ea.retry.fireInstanceId";
    static final String KEY_RETRY_RECOVERING = "ea.retry.recovering";
    static final String PARAMETER_PREFIX = "ea.parameter.";
    static final String IMMEDIATE_PARAMETER_PREFIX = "ea.immediateParameter.";
    static final String GROUP_PREFIX = "ea.scheduler.";

    /** 禁止实例化映射器。 */
    private QuartzScheduleMapper() {
    }

    /**
     * 映射 Quartz Job Key。
     *
     * @param scheduleId 调度标识
     * @return Job Key
     */
    static JobKey jobKey(ScheduleId scheduleId) {
        return new JobKey(scheduleId.name(), GROUP_PREFIX + scheduleId.namespace());
    }

    /**
     * 映射 Quartz Trigger Key。
     *
     * @param scheduleId 调度标识
     * @return Trigger Key
     */
    static TriggerKey triggerKey(ScheduleId scheduleId) {
        return new TriggerKey(scheduleId.name(), GROUP_PREFIX + scheduleId.namespace());
    }

    /**
     * 将公共定义映射为持久 Job。
     *
     * @param definition 调度定义
     * @return Job 详情
     */
    static JobDetail toJobDetail(ScheduleDefinition definition) {
        Class<? extends org.quartz.Job> jobClass =
            definition.concurrencyPolicy() == ConcurrencyPolicy.DISALLOW
                ? DisallowConcurrentDispatchJob.class
                : ConcurrentDispatchJob.class;
        return JobBuilder.newJob(jobClass)
            .withIdentity(jobKey(definition.id()))
            .withDescription(emptyToNull(definition.description()))
            .storeDurably(true)
            .requestRecovery(definition.recoverOnNodeFailure())
            .usingJobData(toJobData(definition))
            .build();
    }

    /**
     * 将公共定义映射为主 Trigger。
     *
     * @param definition 调度定义
     * @return Quartz Trigger
     */
    static Trigger toTrigger(ScheduleDefinition definition) {
        TriggerBuilder<Trigger> builder = TriggerBuilder.newTrigger()
            .withIdentity(triggerKey(definition.id()))
            .forJob(jobKey(definition.id()))
            .usingJobData(identityData(definition.id()));
        SchedulePlan plan = definition.plan();
        if (plan instanceof CronSchedulePlan cronPlan) {
            CronScheduleBuilder schedule = CronScheduleBuilder
                .cronSchedule(cronPlan.expression())
                .inTimeZone(TimeZone.getTimeZone(cronPlan.zoneId()));
            schedule = definition.misfirePolicy() == MisfirePolicy.SKIP
                ? schedule.withMisfireHandlingInstructionDoNothing()
                : schedule.withMisfireHandlingInstructionFireAndProceed();
            return builder.withSchedule(schedule).build();
        }
        if (plan instanceof OnceSchedulePlan oncePlan) {
            SimpleScheduleBuilder schedule = SimpleScheduleBuilder.simpleSchedule()
                .withRepeatCount(0);
            schedule = definition.misfirePolicy() == MisfirePolicy.SKIP
                ? schedule.withMisfireHandlingInstructionNextWithExistingCount()
                : schedule.withMisfireHandlingInstructionFireNow();
            return builder
                .startAt(oncePlan.fireAt())
                .withSchedule(schedule)
                .build();
        }
        throw new ScheduleException(
            ScheduleErrorCode.INVALID_DEFINITION,
            "unsupported schedule plan: " + plan.getClass().getName()
        );
    }

    /**
     * 构造仅对本次立即触发生效的数据。
     *
     * @param scheduleId 调度标识
     * @param invocationId 调用标识
     * @param parameters 已合并并校验的参数快照
     * @return Trigger 数据
     */
    static JobDataMap immediateData(
        ScheduleId scheduleId,
        String invocationId,
        Map<String, String> parameters
    ) {
        JobDataMap data = identityData(scheduleId);
        data.put(KEY_INVOCATION, invocationId);
        data.put(KEY_IMMEDIATE_PARAMETER_SNAPSHOT, Boolean.TRUE.toString());
        putParameters(data, parameters, IMMEDIATE_PARAMETER_PREFIX);
        return data;
    }

    /**
     * 固化持久化重试所需的原始 fire 上下文。
     */
    static JobDataMap retryData(ScheduleFireContext context, int attempt) {
        JobDataMap data = identityData(context.scheduleId());
        // Trigger 数据覆盖 JobDetail；任务定义被 replace 后，既有 fire 的重试仍须
        // 派发给首次触发时的 Handler，而不是意外切换到新 Handler。
        data.put(KEY_HANDLER, context.handlerCode());
        data.put(KEY_RETRY_ATTEMPT, Integer.toString(attempt));
        data.put(KEY_RETRY_SCHEDULED_FIRE_TIME,
            Long.toString(context.scheduledFireTime().toEpochMilli()));
        data.put(KEY_RETRY_ACTUAL_FIRE_TIME,
            Long.toString(context.actualFireTime().toEpochMilli()));
        data.put(KEY_RETRY_FIRE_INSTANCE_ID, context.fireInstanceId());
        data.put(KEY_RETRY_RECOVERING, Boolean.toString(context.recovering()));
        if (context.invocationId() != null) {
            data.put(KEY_INVOCATION, context.invocationId());
        }
        // 重试必须沿用首次 fire 的参数快照，不能读取期间被替换的新定义参数。
        data.put(KEY_IMMEDIATE_PARAMETER_SNAPSHOT, Boolean.TRUE.toString());
        putParameters(data, context.parameters(), IMMEDIATE_PARAMETER_PREFIX);
        return data;
    }

    /**
     * 从持久 JobData 恢复公共定义。
     *
     * @param data 持久数据
     * @return 调度定义
     */
    static ScheduleDefinition toDefinition(JobDataMap data) {
        try {
            String version = required(data, KEY_SCHEMA_VERSION);
            if (!SCHEMA_VERSION.equals(version)) {
                throw new ScheduleException(
                    ScheduleErrorCode.PROVIDER_FAILURE,
                    "unsupported persisted schedule schema version: " + version
                );
            }
            ScheduleId scheduleId = toScheduleId(data);
            SchedulePlan plan = switch (required(data, KEY_PLAN_TYPE)) {
                case "CRON" -> new CronSchedulePlan(
                    required(data, KEY_CRON),
                    ZoneId.of(required(data, KEY_ZONE))
                );
                case "ONCE" -> new OnceSchedulePlan(
                    Instant.ofEpochMilli(Long.parseLong(required(data, KEY_FIRE_AT)))
                );
                default -> throw new IllegalArgumentException("unknown plan type");
            };
            return new ScheduleDefinition(
                scheduleId,
                required(data, KEY_HANDLER),
                plan,
                MisfirePolicy.valueOf(required(data, KEY_MISFIRE)),
                ConcurrencyPolicy.valueOf(required(data, KEY_CONCURRENCY)),
                strictBoolean(data, KEY_RECOVER),
                parameters(data),
                stringValue(data, KEY_DESCRIPTION)
            );
        } catch (ScheduleException exception) {
            if (exception.errorCode() == ScheduleErrorCode.PROVIDER_FAILURE) {
                throw exception;
            }
            throw new ScheduleException(
                ScheduleErrorCode.PROVIDER_FAILURE,
                "invalid persisted schedule data",
                exception
            );
        } catch (RuntimeException exception) {
            throw new ScheduleException(
                ScheduleErrorCode.PROVIDER_FAILURE,
                "invalid persisted schedule data",
                exception
            );
        }
    }

    /**
     * 从 JobData 恢复调度标识。
     *
     * @param data 持久数据
     * @return 调度标识
     */
    static ScheduleId toScheduleId(JobDataMap data) {
        return new ScheduleId(required(data, KEY_NAMESPACE), required(data, KEY_NAME));
    }

    /**
     * 解码 JobData 中的业务参数。
     *
     * @param data 合并后的 JobData
     * @return 不可变参数
     */
    static Map<String, String> parameters(JobDataMap data) {
        String parameterPrefix = immediateParameterSnapshot(data)
            ? IMMEDIATE_PARAMETER_PREFIX
            : PARAMETER_PREFIX;
        Map<String, String> parameters = new HashMap<>();
        for (String key : data.getKeys()) {
            if (!key.startsWith(parameterPrefix)) {
                continue;
            }
            String encodedKey = key.substring(parameterPrefix.length());
            String parameterKey = new String(
                Base64.getUrlDecoder().decode(encodedKey),
                StandardCharsets.UTF_8
            );
            parameters.put(parameterKey, stringValue(data, key));
        }
        return ScheduleDefinition.copyAndValidateParameters(parameters);
    }

    /**
     * 将公共定义编码为纯字符串 JobData。
     *
     * @param definition 调度定义
     * @return 持久数据
     */
    private static JobDataMap toJobData(ScheduleDefinition definition) {
        JobDataMap data = identityData(definition.id());
        data.put(KEY_SCHEMA_VERSION, SCHEMA_VERSION);
        data.put(KEY_HANDLER, definition.handlerCode());
        data.put(KEY_MISFIRE, definition.misfirePolicy().name());
        data.put(KEY_CONCURRENCY, definition.concurrencyPolicy().name());
        data.put(KEY_RECOVER, Boolean.toString(definition.recoverOnNodeFailure()));
        data.put(KEY_DESCRIPTION, definition.description());
        if (definition.plan() instanceof CronSchedulePlan cronPlan) {
            data.put(KEY_PLAN_TYPE, "CRON");
            data.put(KEY_CRON, cronPlan.expression());
            data.put(KEY_ZONE, cronPlan.zoneId().getId());
        } else if (definition.plan() instanceof OnceSchedulePlan oncePlan) {
            data.put(KEY_PLAN_TYPE, "ONCE");
            data.put(KEY_FIRE_AT, Long.toString(oncePlan.fireAt().toEpochMilli()));
        }
        putParameters(data, definition.parameters(), PARAMETER_PREFIX);
        return data;
    }

    /**
     * 构造调度标识数据。
     *
     * @param scheduleId 调度标识
     * @return 标识数据
     */
    private static JobDataMap identityData(ScheduleId scheduleId) {
        JobDataMap data = new JobDataMap();
        data.put(KEY_NAMESPACE, scheduleId.namespace());
        data.put(KEY_NAME, scheduleId.name());
        return data;
    }

    /**
     * 将参数编码到 JobData。
     *
     * @param data 目标 JobData
     * @param parameters 参数
     * @param parameterPrefix 参数键前缀
     */
    private static void putParameters(
        JobDataMap data,
        Map<String, String> parameters,
        String parameterPrefix
    ) {
        for (Map.Entry<String, String> entry : parameters.entrySet()) {
            String encodedKey = Base64.getUrlEncoder().withoutPadding().encodeToString(
                entry.getKey().getBytes(StandardCharsets.UTF_8)
            );
            data.put(parameterPrefix + encodedKey, entry.getValue());
        }
    }

    /**
     * 判断当前数据是否包含立即触发参数快照。
     *
     * @param data 合并后的 JobData
     * @return 存在完整快照时为 true
     */
    private static boolean immediateParameterSnapshot(JobDataMap data) {
        Object marker = data.get(KEY_IMMEDIATE_PARAMETER_SNAPSHOT);
        if (marker == null) {
            return false;
        }
        if (Boolean.TRUE.toString().equals(marker.toString())) {
            return true;
        }
        throw new ScheduleException(
            ScheduleErrorCode.PROVIDER_FAILURE,
            "invalid immediate parameter snapshot marker"
        );
    }

    /**
     * 读取必填字符串字段。
     *
     * @param data JobData
     * @param key 字段键
     * @return 非空字段值
     */
    private static String required(JobDataMap data, String key) {
        String value = stringValue(data, key);
        if (value == null || value.isBlank()) {
            throw new IllegalArgumentException("missing job data: " + key);
        }
        return value;
    }

    /**
     * 将 JobData 字段读取为字符串。
     *
     * @param data JobData
     * @param key 字段键
     * @return 字符串值，可为空
     */
    private static String stringValue(JobDataMap data, String key) {
        Object value = data.get(key);
        return value == null ? null : value.toString();
    }

    /**
     * 严格读取布尔字符串，避免损坏值静默降级为 false。
     *
     * @param data JobData
     * @param key 字段键
     * @return 布尔值
     */
    private static boolean strictBoolean(JobDataMap data, String key) {
        String value = required(data, key);
        if ("true".equals(value)) {
            return true;
        }
        if ("false".equals(value)) {
            return false;
        }
        throw new IllegalArgumentException("invalid boolean job data: " + key);
    }

    /**
     * 将空字符串转换为空值。
     *
     * @param value 原始值
     * @return 非空文本或 null
     */
    private static String emptyToNull(String value) {
        return value == null || value.isEmpty() ? null : value;
    }
}
