package com.dji.sample.wayline.service.impl;

import com.dji.sample.common.error.CommonErrorEnum;
import com.dji.sample.common.model.CustomClaim;
import com.dji.sample.component.mqtt.model.EventsReceiver;
import com.dji.sample.component.redis.RedisConst;
import com.dji.sample.component.redis.RedisOpsUtils;
import com.dji.sample.component.websocket.service.IWebSocketMessageService;
import com.dji.sample.mail.util.MailTemplateUtil;
import com.dji.sample.mail.util.MailUtil;
import com.dji.sample.manage.model.dto.DeviceDTO;
import com.dji.sample.manage.service.IDeviceRedisService;
import com.dji.sample.media.model.MediaFileCountDTO;
import com.dji.sample.media.service.IMediaRedisService;
import com.dji.sample.wayline.model.dto.ConditionalWaylineJobKey;
import com.dji.sample.wayline.model.dto.WaylineJobDTO;
import com.dji.sample.wayline.model.dto.WaylineTaskConditionDTO;
import com.dji.sample.wayline.model.entity.WaylineJobEntity;
import com.dji.sample.wayline.model.enums.WaylineErrorCodeEnum;
import com.dji.sample.wayline.model.enums.WaylineJobStatusEnum;
import com.dji.sample.wayline.model.param.BreakPointWaylineJobEvent;
import com.dji.sample.wayline.model.param.CreateInFlightJobParam;
import com.dji.sample.wayline.model.param.CreateJobParam;
import com.dji.sample.wayline.model.param.UpdateJobParam;
import com.dji.sample.wayline.service.*;
import com.dji.sdk.cloudapi.device.*;
import com.dji.sdk.cloudapi.media.UploadFlighttaskMediaPrioritize;
import com.dji.sdk.cloudapi.media.api.AbstractMediaService;
import com.dji.sdk.cloudapi.wayline.*;
import com.dji.sdk.cloudapi.wayline.api.AbstractWaylineService;
import com.dji.sdk.common.BaseModel;
import com.dji.sdk.common.HttpResultResponse;
import com.dji.sdk.common.SDKManager;
import com.dji.sdk.exception.CloudSDKErrorEnum;
import com.dji.sdk.exception.CloudSDKException;
import com.dji.sdk.mqtt.MqttReply;
import com.dji.sdk.mqtt.events.TopicEventsRequest;
import com.dji.sdk.mqtt.events.TopicEventsResponse;
import com.dji.sdk.mqtt.services.ServicesReplyData;
import com.dji.sdk.mqtt.services.TopicServicesResponse;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.messaging.MessageHeaders;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.context.event.EventListener;

import java.net.URL;
import java.sql.SQLException;
import java.time.*;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @author sean
 * @version 1.1
 * @date 2022/6/9
 */
@Service
@Slf4j
public class FlightTaskServiceImpl extends AbstractWaylineService implements IFlightTaskService {

    @Autowired
    private ObjectMapper mapper;

    @Autowired
    private IWebSocketMessageService websocketMessageService;

    @Autowired
    private IWaylineJobService waylineJobService;

    @Autowired
    private IDeviceRedisService deviceRedisService;

    @Autowired
    private IWaylineRedisService waylineRedisService;

    @Autowired
    private IMediaRedisService mediaRedisService;

    @Autowired
    private IWaylineFileService waylineFileService;

    @Autowired
    private SDKWaylineService abstractWaylineService;

    @Autowired
    @Qualifier("mediaServiceImpl")
    private AbstractMediaService abstractMediaService;

    @Autowired
    private MailUtil mailUtil;

    @Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS)
    public void checkScheduledJob() {
        Object jobIdValue = RedisOpsUtils.zGetMin(RedisConst.WAYLINE_JOB_TIMED_EXECUTE);
        if (Objects.isNull(jobIdValue)) {
            return;
        }
        log.info("Check the timed tasks of the wayline. {}", jobIdValue);
        // format: {workspace_id}:{dock_sn}:{job_id}
        String[] jobArr = String.valueOf(jobIdValue).split(RedisConst.DELIMITER);
        double time = RedisOpsUtils.zScore(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
        long now = System.currentTimeMillis();
        int offset = 30_000;

        // Expired tasks are deleted directly.
        if (time < now - offset) {
            RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .jobId(jobArr[2])
                    .status(WaylineJobStatusEnum.FAILED.getVal())
                    .executeTime(LocalDateTime.now())
                    .completedTime(LocalDateTime.now())
                    .code(HttpStatus.SC_REQUEST_TIMEOUT).build());
            return;
        }

        if (now <= time && time <= now + offset) {
            try {
                this.executeFlightTask(jobArr[0], jobArr[2]);
            } catch (Exception e) {
                log.info("The scheduled task delivery failed.");
                waylineJobService.updateJob(WaylineJobDTO.builder()
                        .jobId(jobArr[2])
                        .status(WaylineJobStatusEnum.FAILED.getVal())
                        .executeTime(LocalDateTime.now())
                        .completedTime(LocalDateTime.now())
                        .code(HttpStatus.SC_INTERNAL_SERVER_ERROR).build());
            } finally {
                RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
            }
        }
    }

    @Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS)
    public void prepareConditionJob() {
        Optional<ConditionalWaylineJobKey> jobKeyOpt = waylineRedisService.getNearestConditionalWaylineJob();
        if (jobKeyOpt.isEmpty()) {
            return;
        }
        ConditionalWaylineJobKey jobKey = jobKeyOpt.get();
        log.info("Check the conditional tasks of the wayline. {}", jobKey.toString());
        // format: {workspace_id}:{dock_sn}:{job_id}
        double time = waylineRedisService.getConditionalWaylineJobTime(jobKey);
        long now = System.currentTimeMillis();
        // prepare the task one day in advance.
        int offset = 86_400_000;

        if (now + offset < time) {
            return;
        }

        WaylineJobDTO job = WaylineJobDTO.builder()
                .jobId(jobKey.getJobId())
                .status(WaylineJobStatusEnum.FAILED.getVal())
                .executeTime(LocalDateTime.now())
                .completedTime(LocalDateTime.now())
                .code(HttpStatus.SC_INTERNAL_SERVER_ERROR).build();
        try {
            Optional<WaylineJobDTO> waylineJobOpt = waylineRedisService.getConditionalWaylineJob(jobKey.getJobId());
            if (waylineJobOpt.isEmpty()) {
                job.setCode(CommonErrorEnum.REDIS_DATA_NOT_FOUND.getCode());
                waylineJobService.updateJob(job);
                waylineRedisService.removePrepareConditionalWaylineJob(jobKey);
                return;
            }
            WaylineJobDTO waylineJob = waylineJobOpt.get();

            HttpResultResponse result = this.publishOneFlightTask(waylineJob);
            waylineRedisService.removePrepareConditionalWaylineJob(jobKey);

            if (HttpResultResponse.CODE_SUCCESS == result.getCode()) {
                return;
            }

            // If the end time is exceeded, no more retries will be made.
            waylineRedisService.delConditionalWaylineJob(jobKey.getJobId());
            if (waylineJob.getEndTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() - RedisConst.WAYLINE_JOB_BLOCK_TIME * 1000 < now) {
                return;
            }

            // Retry if the end time has not been exceeded.
            this.retryPrepareJob(jobKey, waylineJob);

        } catch (Exception e) {
            log.info("Failed to prepare the conditional task.");
            waylineJobService.updateJob(job);
        }

    }

    /**
     * For immediate tasks, the server time shall prevail.
     * @param param
     */
    private void fillImmediateTime(CreateJobParam param) {
        if (TaskTypeEnum.IMMEDIATE != param.getTaskType()) {
            return;
        }
        long now = System.currentTimeMillis() / 1000;
        param.setTaskDays(List.of(now));
        param.setTaskPeriods(List.of(List.of(now)));
    }


    private void addConditions(WaylineJobDTO waylineJob, CreateJobParam param, Long beginTime, Long endTime) {
        if (TaskTypeEnum.CONDITIONAL != param.getTaskType()) {
            return;
        }

        waylineJob.setConditions(
                WaylineTaskConditionDTO.builder()
                        .executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ?
                                new ExecutableConditions().setStorageCapacity(param.getMinStorageCapacity()) : null)
                        .readyConditions(new ReadyConditions()
                                .setBatteryCapacity(param.getMinBatteryCapacity())
                                .setBeginTime(beginTime)
                                .setEndTime(endTime))
                        .build());

        waylineRedisService.setConditionalWaylineJob(waylineJob);
        // key: wayline_job_condition, value: {workspace_id}:{dock_sn}:{job_id}
        boolean isAdd = waylineRedisService.addPrepareConditionalWaylineJob(waylineJob);
        if (!isAdd) {
            throw new RuntimeException("Failed to create conditional job.");
        }
    }

    @Override
    public HttpResultResponse publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException {
        fillImmediateTime(param);

        for (Long taskDay : param.getTaskDays()) {
            LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(taskDay), ZoneId.systemDefault());
            for (List<Long> taskPeriod : param.getTaskPeriods()) {
                long beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(0)), ZoneId.systemDefault()))
                        .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                long endTime = taskPeriod.size() > 1 ?
                        LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault()))
                                .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;
                if (TaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
                    continue;
                }

                Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.createWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime);
                if (waylineJobOpt.isEmpty()) {
                    throw new SQLException("Failed to create wayline job.");
                }
                WaylineJobDTO waylineJob = waylineJobOpt.get();
                /**
                 * 起始飞行断点
                 */
                if (param.getBreakPoint() != null) {
                    waylineJob.setBreakPoint(param.getBreakPoint());
                }
                /**
                 * 模拟飞行
                 */
                if (param.getSimulateMission() != null) {
                    waylineJob.setSimulateMission(param.getSimulateMission());
                }
                // If it is a conditional task type, add conditions to the job parameters.
                addConditions(waylineJob, param, beginTime, endTime);

                HttpResultResponse response = this.publishOneFlightTask(waylineJob);
                if (HttpResultResponse.CODE_SUCCESS != response.getCode()) {
                    return response;
                }
            }
        }
        return HttpResultResponse.success();
    }

    @SneakyThrows
    @EventListener
    public HttpResultResponse publishTimeFlightTask(BreakPointWaylineJobEvent event) {

        String taskId = event.getTaskId();
        FlighttaskBreakPoint breakPoint = event.getBreakPointInfo();
        return this.publishTimeFlightTask(taskId, breakPoint);
    }

    @SneakyThrows
    @Override
    public HttpResultResponse publishTimeFlightTask(String taskId, FlighttaskBreakPoint breakPoint) {
        // 查询任务详情
        WaylineJobEntity jobEntity = waylineJobService.getJobEntityByJobId(taskId);

        // token
        CustomClaim customClaim = new CustomClaim();
        customClaim.setWorkspaceId(jobEntity.getWorkspaceId());

        customClaim.setUsername(jobEntity.getUsername());
        customClaim.setOrgId(jobEntity.getOrgId());

        // 创建 开始和结束时间是 20-40分钟 电量90 条件任务
        CreateJobParam param = new CreateJobParam();
        // 任务信息
        param.setOrgId(jobEntity.getOrgId());
        String breakPointTaskName = getBreakPointTaskName(jobEntity.getName());
        param.setName(breakPointTaskName);

        param.setFileId(jobEntity.getFileId());
        param.setDockSn(jobEntity.getDockSn());
        param.setWaylineType(WaylineTypeEnum.find(jobEntity.getWaylineType()));
        param.setRthAltitude(jobEntity.getRthAltitude());
        param.setOutOfControlAction(OutOfControlActionEnum.find(jobEntity.getOutOfControlAction()));

        // 断点
        param.setBreakPoint(breakPoint);

        // 条件信息
        param.setTaskType(TaskTypeEnum.CONDITIONAL);
        param.setMinBatteryCapacity(90);
        ZoneId zone = ZoneId.systemDefault();
        // 当前时间加20分钟
        ZonedDateTime now = ZonedDateTime.now(zone);
        ZonedDateTime plus30Min = now.plusMinutes(20);
        long timestampPlus30Min = plus30Min.toInstant().getEpochSecond();
        // 20分钟之后当天0点（如果跨天就是第二天0点）
        LocalDateTime targetDate = plus30Min.toLocalDate().atStartOfDay();
        long timestampMidnight = targetDate.atZone(zone).toInstant().getEpochSecond();
        List<Long> taskDays = new ArrayList<>();
        taskDays.add(timestampMidnight);
        param.setTaskDays(taskDays);
        List<List<Long>> taskPeriods = new ArrayList<>();
        List<Long> tempPeriod = new ArrayList<>();
        tempPeriod.add(timestampPlus30Min);
        // 结束时间 再加20分钟
        tempPeriod.add(timestampPlus30Min + 20 * 60);
        taskPeriods.add(tempPeriod);
        param.setTaskPeriods(taskPeriods);
        // 自动断点续飞
        param.setBreakPointResumeFlight(1);

        // minStorageCapacity   存储容量

        return this.publishFlightTask(param, customClaim);
    }

    public static String getBreakPointTaskName(String name) {
        if (name == null || name.isEmpty()) {
            return name;
        }
        // 设置 断点续飞任务名称规则 -x
        Pattern pattern = Pattern.compile("^(.*?)-(\\d+)$");
        Matcher matcher = pattern.matcher(name);

        if (matcher.matches()) {
            String prefix = matcher.group(1);
            int num = Integer.parseInt(matcher.group(2));
            return prefix + "-" + (num + 1);
        } else {
            return name + "-1";
        }
    }

    /**
     * 校验网关可下发航线：机场走 Dock mode_code；Pilot RC「模拟机场」仅要求 RC 与子机在线。
     */
    private void assertGatewayReadyForWayline(String gatewaySn) {
        Optional<DeviceDTO> gatewayOpt = deviceRedisService.getDeviceOnline(gatewaySn);
        if (gatewayOpt.isEmpty()) {
            throw new RuntimeException("Gateway is offline.");
        }
        DeviceDTO gateway = gatewayOpt.get();
        if (DeviceDomainEnum.REMOTER_CONTROL == gateway.getDomain()) {
            String childSn = gateway.getChildDeviceSn();
            if (!StringUtils.hasText(childSn) || !deviceRedisService.checkDeviceOnline(childSn)) {
                throw new RuntimeException("Aircraft under RC is offline.");
            }
            return;
        }
        DockModeCodeEnum dockModeCodeEnum = deviceRedisService.getDeviceOsd(gatewaySn, OsdDock.class)
                .map(OsdDock::getModeCode).orElse(null);
        if (dockModeCodeEnum == null || dockModeCodeEnum == DockModeCodeEnum.REMOTE_DEBUGGING) {
            throw new RuntimeException("Dock is remote_debugging state, does not support flight task.");
        }
    }

    private boolean isRemoterControlGateway(String gatewaySn) {
        return deviceRedisService.getDeviceOnline(gatewaySn)
                .map(d -> DeviceDomainEnum.REMOTER_CONTROL == d.getDomain())
                .orElse(false);
    }

    private boolean isMqttNoReply(CloudSDKException e) {
        return e.getErrorInfo() != null
                && Objects.equals(CloudSDKErrorEnum.MQTT_PUBLISH_ABNORMAL.getCode(), e.getErrorInfo().getCode());
    }

    private String mqttNoReplyHint(String gatewaySn, String method) {
        if (isRemoterControlGateway(gatewaySn)) {
            return "Pilot 遥控器未应答 " + method + "（211001）。"
                    + "官方 Pilot 航线管理为 HTTPS 文件库；向 RC 下发的 flighttask_* 需固件支持。"
                    + "请确认 dock_sn 为遥控器 SN，并用 MQTT 抓包核对 thing/product/{rc_sn}/services 与 services_reply。";
        }
        return "设备未应答 " + method + "（211001）。请确认网关在线且已订阅 services，检查 MQTT 连通与 SDKManager 注册。";
    }

    private void markJobMqttFailed(WaylineJobDTO waylineJob, CloudSDKException e) {
        waylineJobService.updateJob(WaylineJobDTO.builder()
                .workspaceId(waylineJob.getWorkspaceId())
                .jobId(waylineJob.getJobId())
                .executeTime(LocalDateTime.now())
                .status(WaylineJobStatusEnum.FAILED.getVal())
                .completedTime(LocalDateTime.now())
                .code(e.getErrorInfo() != null ? e.getErrorInfo().getCode() : CloudSDKErrorEnum.MQTT_PUBLISH_ABNORMAL.getCode())
                .build());
    }

    public HttpResultResponse publishOneFlightTask(WaylineJobDTO waylineJob) throws SQLException {

        // 实测：当前 Pilot 固件对 flighttask_* 不回 services_reply（211001）。
        // 官方 Pilot 航线为 HTTPS 文件库（App 内 Mission），勿空等 MQTT 超时。
        if (isRemoterControlGateway(waylineJob.getDockSn())) {
            String msg = "Pilot 上云暂不支持 Web 端 MQTT 航线任务下发（flighttask_prepare）。"
                    + "请在 DJI Pilot 2 航线库中执行航线，或改用机场设备创建任务。";
            log.warn("Reject wayline job {} for Pilot RC {}: {}", waylineJob.getJobId(), waylineJob.getDockSn(), msg);
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .workspaceId(waylineJob.getWorkspaceId())
                    .jobId(waylineJob.getJobId())
                    .executeTime(LocalDateTime.now())
                    .status(WaylineJobStatusEnum.FAILED.getVal())
                    .completedTime(LocalDateTime.now())
                    .code(CloudSDKErrorEnum.MQTT_PUBLISH_ABNORMAL.getCode())
                    .build());
            return HttpResultResponse.error(msg);
        }

        assertGatewayReadyForWayline(waylineJob.getDockSn());

        boolean isSuccess = this.prepareFlightTask(waylineJob);
        if (isSuccess) {
            // 任务准备成功
            log.info("Prepare job successfully.{}", waylineJob.getJobId());
        } else {
            return HttpResultResponse.error("Failed to prepare job.");
        }

        // Issue an immediate task execution command.
        if (TaskTypeEnum.IMMEDIATE == waylineJob.getTaskType()) {
            if (!executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId())) {
                return HttpResultResponse.error("Failed to execute job.");
            }
        }

        if (TaskTypeEnum.TIMED == waylineJob.getTaskType()) {
            // key: wayline_job_timed, value: {workspace_id}:{dock_sn}:{job_id}
            boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE,
                    waylineJob.getWorkspaceId() + RedisConst.DELIMITER + waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(),
                    waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
            if (!isAdd) {
                return HttpResultResponse.error("Failed to create scheduled job.");
            }
        }
        // 发送 邮件
        this.sendCreatedTaskMail(waylineJob);

        return HttpResultResponse.success();
    }

    @Async
    public void sendCreatedTaskMail(WaylineJobDTO waylineJob) {

        // 获取当前 设备 相关人员
        String[] emailSender = MailTemplateUtil.getSendTo();
        // 获取邮件
        String subject = "Geofly task created email";
        // 生成邮件数据
        String content = MailTemplateUtil.buildTaskCreatedTemplate(waylineJob.getJobName(), waylineJob.getUsername());
        // 发送邮件
        for (String to : emailSender) {
            mailUtil.sendMail(to, subject, content);
        }

    }

    private Boolean prepareFlightTask(WaylineJobDTO waylineJob) throws SQLException {
        // get wayline file
        Optional<GetWaylineListResponse> waylineFile = waylineFileService.getWaylineByWaylineId(waylineJob.getWorkspaceId(), waylineJob.getFileId());
        if (waylineFile.isEmpty()) {
            throw new SQLException("Wayline file doesn't exist.");
        }

        // get file url
        URL url = waylineFileService.getObjectUrl(waylineJob.getWorkspaceId(), waylineFile.get().getId());

        FlighttaskPrepareRequest flightTask = new FlighttaskPrepareRequest()
                .setFlightId(waylineJob.getJobId())
                .setExecuteTime(waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
                .setTaskType(waylineJob.getTaskType())
                .setWaylineType(waylineJob.getWaylineType())
                .setRthAltitude(waylineJob.getRthAltitude())
                .setOutOfControlAction(waylineJob.getOutOfControlAction())
                .setExitWaylineWhenRcLost(ExitWaylineWhenRcLostEnum.EXECUTE_RC_LOST_ACTION)
                .setWaylinePrecisionType(waylineJob.getWaylinePrecisionType() != null ? waylineJob.getWaylinePrecisionType() : WaylinePrecisionTypeEnum.RTK)
                .setFile(new FlighttaskFile()
                        .setUrl(url.toString())
                        .setFingerprint(waylineFile.get().getSign()));
        if (waylineJob.getCustomProcess() != null) {
            flightTask.setCustomProcess(waylineJob.getCustomProcess());
        }
        if (waylineJob.getSimulateMission() != null) {
            flightTask.setSimulateMission(waylineJob.getSimulateMission());
        }
        // 设置航线任务断点
        if (waylineJob.getBreakPoint() != null) {
            flightTask.setBreakPoint(waylineJob.getBreakPoint());
        }

        if (TaskTypeEnum.CONDITIONAL == waylineJob.getTaskType()) {
            if (Objects.isNull(waylineJob.getConditions())) {
                throw new IllegalArgumentException();
            }
            flightTask.setReadyConditions(waylineJob.getConditions().getReadyConditions());
            flightTask.setExecutableConditions(waylineJob.getConditions().getExecutableConditions());
        }

        TopicServicesResponse<ServicesReplyData> serviceReply;
        try {
            serviceReply = abstractWaylineService.flighttaskPrepare(
                    SDKManager.getDeviceSDK(waylineJob.getDockSn()), flightTask);
        } catch (CloudSDKException e) {
            markJobMqttFailed(waylineJob, e);
            throw new RuntimeException(mqttNoReplyHint(waylineJob.getDockSn(), "flighttask_prepare"), e);
        }
        if (!serviceReply.getData().getResult().isSuccess()) {
            log.info("Prepare task ====> Error code: {}", serviceReply.getData().getResult());
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .workspaceId(waylineJob.getWorkspaceId())
                    .jobId(waylineJob.getJobId())
                    .executeTime(LocalDateTime.now())
                    .status(WaylineJobStatusEnum.FAILED.getVal())
                    .completedTime(LocalDateTime.now())
                    .code(serviceReply.getData().getResult().getCode()).build());
            return false;
        }
        return true;
    }


    @Override
    public Boolean executeFlightTask(String workspaceId, String jobId) {
        // get job
        Optional<WaylineJobDTO> waylineJob = waylineJobService.getJobByJobId(workspaceId, jobId);
        if (waylineJob.isEmpty()) {
            throw new IllegalArgumentException("Job doesn't exist.");
        }

        boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.get().getDockSn());
        if (!isOnline) {
            throw new RuntimeException("Dock is offline.");
        }

        WaylineJobDTO job = waylineJob.get();

        TopicServicesResponse<ServicesReplyData> serviceReply;
        try {
            serviceReply = abstractWaylineService.flighttaskExecute(
                    SDKManager.getDeviceSDK(job.getDockSn()), new FlighttaskExecuteRequest().setFlightId(jobId));
        } catch (CloudSDKException e) {
            throw new RuntimeException(mqttNoReplyHint(job.getDockSn(), "flighttask_execute"), e);
        }
        if (!serviceReply.getData().getResult().isSuccess()) {
            log.info("Execute job ====> Error: {}", serviceReply.getData().getResult());
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .jobId(jobId)
                    .executeTime(LocalDateTime.now())
                    .status(WaylineJobStatusEnum.FAILED.getVal())
                    .completedTime(LocalDateTime.now())
                    .code(serviceReply.getData().getResult().getCode()).build());
            // The conditional task fails and enters the blocking status.
            if (TaskTypeEnum.CONDITIONAL == job.getTaskType()
                    && WaylineErrorCodeEnum.find(serviceReply.getData().getResult().getCode()).isBlock()) {
                waylineRedisService.setBlockedWaylineJob(job.getDockSn(), jobId);
            }
            return false;
        }

        waylineJobService.updateJob(WaylineJobDTO.builder()
                .jobId(jobId)
                .executeTime(LocalDateTime.now())
                .status(WaylineJobStatusEnum.IN_PROGRESS.getVal())
                .build());
        waylineRedisService.setRunningWaylineJob(job.getDockSn(), EventsReceiver.<FlighttaskProgress>builder().bid(jobId).sn(job.getDockSn()).build());
        return true;
    }

    @Override
    public void cancelFlightTask(String workspaceId, Collection<String> jobIds) {
        List<WaylineJobDTO> waylineJobs = waylineJobService.getJobsByConditions(workspaceId, jobIds, WaylineJobStatusEnum.PENDING);

        Set<String> waylineJobIds = waylineJobs.stream().map(WaylineJobDTO::getJobId).collect(Collectors.toSet());
        // Check if the task status is correct.
        boolean isErr = !jobIds.removeAll(waylineJobIds) || !jobIds.isEmpty() ;
        if (isErr) {
            throw new IllegalArgumentException("These tasks have an incorrect status and cannot be canceled. " + Arrays.toString(jobIds.toArray()));
        }

        // Group job id by dock sn.
        Map<String, List<String>> dockJobs = waylineJobs.stream()
                .collect(Collectors.groupingBy(WaylineJobDTO::getDockSn,
                        Collectors.mapping(WaylineJobDTO::getJobId, Collectors.toList())));
        dockJobs.forEach((dockSn, idList) -> this.publishCancelTask(workspaceId, dockSn, idList));

    }

    public void publishCancelTask(String workspaceId, String dockSn, List<String> jobIds) {
        boolean isOnline = deviceRedisService.checkDeviceOnline(dockSn);
        if (!isOnline) {
            throw new RuntimeException("Dock is offline.");
        }

        boolean mqttOk = false;
        try {
            TopicServicesResponse<ServicesReplyData> serviceReply = abstractWaylineService.flighttaskUndo(
                    SDKManager.getDeviceSDK(dockSn),
                    new FlighttaskUndoRequest().setFlightIds(jobIds));
            if (!serviceReply.getData().getResult().isSuccess()) {
                log.info("Cancel job ====> Error: {}", serviceReply.getData().getResult());
                throw new RuntimeException("Failed to cancel the wayline job of " + dockSn);
            }
            mqttOk = true;
        } catch (CloudSDKException e) {
            // Pilot RC：PENDING 任务常因设备从不应答 flighttask_*；允许仅落库取消，避免删任务卡死
            if (isMqttNoReply(e) && isRemoterControlGateway(dockSn)) {
                log.warn("flighttask_undo no reply from Pilot RC {}, soft-cancel jobs {} in DB only. {}",
                        dockSn, jobIds, e.getMessage());
            } else {
                throw new RuntimeException(mqttNoReplyHint(dockSn, "flighttask_undo"), e);
            }
        }

        for (String jobId : jobIds) {
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .workspaceId(workspaceId)
                    .jobId(jobId)
                    .status(WaylineJobStatusEnum.CANCEL.getVal())
                    .completedTime(LocalDateTime.now())
                    .build());
            RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, workspaceId + RedisConst.DELIMITER + dockSn + RedisConst.DELIMITER + jobId);
        }
        if (!mqttOk) {
            log.info("Soft-cancelled {} job(s) for Pilot RC {} without device undo ack", jobIds.size(), dockSn);
        }

    }

    @Override
    public void uploadMediaHighestPriority(String workspaceId, String jobId) {
        Optional<WaylineJobDTO> jobOpt = waylineJobService.getJobByJobId(workspaceId, jobId);
        if (jobOpt.isEmpty()) {
            throw new RuntimeException(CommonErrorEnum.ILLEGAL_ARGUMENT.getMessage());
        }

        String dockSn = jobOpt.get().getDockSn();
        String key = RedisConst.MEDIA_HIGHEST_PRIORITY_PREFIX + dockSn;
        if (RedisOpsUtils.checkExist(key) && jobId.equals(((MediaFileCountDTO) RedisOpsUtils.get(key)).getJobId())) {
            return;
        }

        TopicServicesResponse<ServicesReplyData> reply = abstractMediaService.uploadFlighttaskMediaPrioritize(
                SDKManager.getDeviceSDK(dockSn), new UploadFlighttaskMediaPrioritize().setFlightId(jobId));
        if (!reply.getData().getResult().isSuccess()) {
            throw new RuntimeException("Failed to set media job upload priority. Error: " + reply.getData().getResult());
        }
    }

    @Override
    public void updateJobStatus(String workspaceId, String jobId, UpdateJobParam param) {
        Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.getJobByJobId(workspaceId, jobId);
        if (waylineJobOpt.isEmpty()) {
            throw new RuntimeException("The job does not exist.");
        }
        WaylineJobDTO waylineJob = waylineJobOpt.get();
        WaylineJobStatusEnum statusEnum = waylineJobService.getWaylineState(waylineJob.getDockSn());
        if (statusEnum.getEnd() || WaylineJobStatusEnum.PENDING == statusEnum) {
            throw new RuntimeException("The wayline job status does not match, and the operation cannot be performed.");
        }

        switch (param.getStatus()) {
            case PAUSE:
                pauseJob(workspaceId, waylineJob.getDockSn(), jobId, statusEnum);
                break;
            case RESUME:
                resumeJob(workspaceId, waylineJob.getDockSn(), jobId, statusEnum);
                break;
        }

    }

    private void pauseJob(String workspaceId, String dockSn, String jobId, WaylineJobStatusEnum statusEnum) {
        if (WaylineJobStatusEnum.PAUSED == statusEnum && jobId.equals(waylineRedisService.getPausedWaylineJobId(dockSn))) {
            waylineRedisService.setPausedWaylineJob(dockSn, jobId);
            return;
        }

        TopicServicesResponse<ServicesReplyData> reply = abstractWaylineService.flighttaskPause(SDKManager.getDeviceSDK(dockSn));
        if (!reply.getData().getResult().isSuccess()) {
            throw new RuntimeException("Failed to pause wayline job. Error: " + reply.getData().getResult());
        }
        waylineRedisService.delRunningWaylineJob(dockSn);
        waylineRedisService.setPausedWaylineJob(dockSn, jobId);
    }

    private void resumeJob(String workspaceId, String dockSn, String jobId, WaylineJobStatusEnum statusEnum) {
        Optional<EventsReceiver<FlighttaskProgress>> runningDataOpt = waylineRedisService.getRunningWaylineJob(dockSn);
        if (WaylineJobStatusEnum.IN_PROGRESS == statusEnum && jobId.equals(runningDataOpt.map(EventsReceiver::getSn).get())) {
            waylineRedisService.setRunningWaylineJob(dockSn, runningDataOpt.get());
            return;
        }
        TopicServicesResponse<ServicesReplyData> reply = abstractWaylineService.flighttaskRecovery(SDKManager.getDeviceSDK(dockSn));
        if (!reply.getData().getResult().isSuccess()) {
            throw new RuntimeException("Failed to resume wayline job. Error: " + reply.getData().getResult());
        }

        runningDataOpt.ifPresent(runningData -> waylineRedisService.setRunningWaylineJob(dockSn, runningData));
        waylineRedisService.delPausedWaylineJob(dockSn);
    }

    @Override
    public void retryPrepareJob(ConditionalWaylineJobKey jobKey, WaylineJobDTO waylineJob) {
        Optional<WaylineJobDTO> childJobOpt = waylineJobService.createWaylineJobByParent(jobKey.getWorkspaceId(), jobKey.getJobId());
        if (childJobOpt.isEmpty()) {
            log.error("Failed to create wayline job.");
            return;
        }

        WaylineJobDTO newJob = childJobOpt.get();
        newJob.setBeginTime(LocalDateTime.now().plusSeconds(RedisConst.WAYLINE_JOB_BLOCK_TIME));
        boolean isAdd = waylineRedisService.addPrepareConditionalWaylineJob(newJob);
        if (!isAdd) {
            log.error("Failed to create wayline job. {}", newJob.getJobId());
            return;
        }

        waylineJob.setJobId(newJob.getJobId());
        waylineRedisService.setConditionalWaylineJob(waylineJob);
    }


    @Override
    public TopicEventsResponse<MqttReply> flighttaskReady(TopicEventsRequest<FlighttaskReady> response, MessageHeaders headers) {
        List<String> flightIds = response.getData().getFlightIds();

        log.info("ready task list：{}", Arrays.toString(flightIds.toArray()) );
        // Check conditional task blocking status.
        String blockedId = waylineRedisService.getBlockedWaylineJobId(response.getGateway());
        // fix: 修复条件任务不执行bug
        if (StringUtils.hasText(blockedId)) {
            log.info("The conditional task is blocking and will not be executed.block id {}", blockedId);
            return null;
        }

        Optional<DeviceDTO> deviceOpt = deviceRedisService.getDeviceOnline(response.getGateway());
        if (deviceOpt.isEmpty()) {
            log.info("The device is offline.flight task ready!");
            return null;
        }
        DeviceDTO device = deviceOpt.get();

        try {
            for (String jobId : flightIds) {
                boolean isExecute = this.executeFlightTask(device.getWorkspaceId(), jobId);
                if (!isExecute) {
                    return null;
                }
                Optional<WaylineJobDTO> waylineJobOpt = waylineRedisService.getConditionalWaylineJob(jobId);
                if (waylineJobOpt.isEmpty()) {
                    log.info("The conditional job has expired and will no longer be executed.");
                    return new TopicEventsResponse<>();
                }
                WaylineJobDTO waylineJob = waylineJobOpt.get();
                this.retryPrepareJob(new ConditionalWaylineJobKey(device.getWorkspaceId(), response.getGateway(), jobId), waylineJob);
                return new TopicEventsResponse<>();
            }
        } catch (Exception e) {
            log.error("Failed to execute conditional task.");
            e.printStackTrace();
        }
        return new TopicEventsResponse<>();
    }

    @Override
    public HttpResultResponse publishInFlightTask(CreateInFlightJobParam param, CustomClaim customClaim) throws SQLException {

        fillInflightImmediateTime(param);

        // 直接默认 立即起飞
        for (Long taskDay : param.getTaskDays()) {
            LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(taskDay), ZoneId.systemDefault());
            for (List<Long> taskPeriod : param.getTaskPeriods()) {
                long beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(0)), ZoneId.systemDefault()))
                        .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                long endTime = taskPeriod.size() > 1 ?
                        LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault()))
                                .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;
                if (TaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
                    continue;
                }

                Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.createInFlightWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime);
                if (waylineJobOpt.isEmpty()) {
                    throw new SQLException("Failed to create wayline job.");
                }
                WaylineJobDTO waylineJob = waylineJobOpt.get();
                // If it is a conditional task type, add conditions to the job parameters.
                // todo
                addConditionsInFlight(waylineJob, param, beginTime, endTime);
                HttpResultResponse response = this.publishOneInFlightFlightTask(waylineJob);
                if (HttpResultResponse.CODE_SUCCESS != response.getCode()) {
                    return response;
                }
            }
        }

        return HttpResultResponse.success();
    }

    private void fillInflightImmediateTime(CreateInFlightJobParam param) {
        if (TaskTypeEnum.IMMEDIATE != param.getTaskType()) {
            return;
        }
        long now = System.currentTimeMillis() / 1000;
        param.setTaskDays(List.of(now));
        param.setTaskPeriods(List.of(List.of(now)));
    }

    private void addConditionsInFlight(WaylineJobDTO waylineJob, CreateInFlightJobParam param, Long beginTime, Long endTime) {

        if (TaskTypeEnum.CONDITIONAL != param.getTaskType()) {
            return;
        }

        waylineJob.setConditions(
                WaylineTaskConditionDTO.builder()
//                        .executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ?
//                                new ExecutableConditions().setStorageCapacity(param.getMinStorageCapacity()) : null)
                        .executableConditions(null)
                        .readyConditions(new ReadyConditions()
                                // .setBatteryCapacity(param.getMinBatteryCapacity())
                                .setBeginTime(beginTime)
                                .setEndTime(endTime))
                        .build());

        waylineRedisService.setConditionalWaylineJob(waylineJob);
        // key: wayline_job_condition, value: {workspace_id}:{dock_sn}:{job_id}
        boolean isAdd = waylineRedisService.addPrepareConditionalWaylineJob(waylineJob);
        if (!isAdd) {
            throw new RuntimeException("Failed to create conditional job.");
        }
    }

    public HttpResultResponse publishOneInFlightFlightTask(WaylineJobDTO waylineJob) throws SQLException {

        assertGatewayReadyForWayline(waylineJob.getDockSn());

        boolean isSuccess = this.prepareInFlightTask(waylineJob);
        if (!isSuccess) {
            return HttpResultResponse.error("Failed to prepare In flight job.");
        }

        // Issue an immediate task execution command.
        // 不需要执行这个就可以直接起飞
//        if (TaskTypeEnum.IMMEDIATE == waylineJob.getTaskType()) {
//            if (!executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId())) {
//                return HttpResultResponse.error("Failed to execute job.");
//            }
//        }

        // 发送 邮件
        // this.sendCreatedTaskMail(waylineJob);

        return HttpResultResponse.success();
    }

    private Boolean prepareInFlightTask(WaylineJobDTO waylineJob) throws SQLException {
        // get wayline file
        Optional<GetWaylineListResponse> waylineFile = waylineFileService.getWaylineByWaylineId(waylineJob.getWorkspaceId(), waylineJob.getFileId());
        if (waylineFile.isEmpty()) {
            throw new SQLException("Wayline file doesn't exist.");
        }

        // get file url
        URL url = waylineFileService.getObjectUrl(waylineJob.getWorkspaceId(), waylineFile.get().getId());

        InFlightWaylineTaskPrepareRequest flightTask = new InFlightWaylineTaskPrepareRequest()
                .setInFlightWaylineId(waylineJob.getJobId())
                .setRthAltitude(waylineJob.getRthAltitude())
                .setOutOfControlAction(waylineJob.getOutOfControlAction())
                .setExitWaylineWhenRcLost(ExitWaylineWhenRcLostEnum.EXECUTE_RC_LOST_ACTION)
                .setWaylinePrecisionType(WaylinePrecisionTypeEnum.RTK)
                .setFile(new FlighttaskFile()
                        .setUrl(url.toString())
                        .setFingerprint(waylineFile.get().getSign()));
        if (waylineJob.getCustomProcess() != null) {
            flightTask.setCustomProcess(waylineJob.getCustomProcess());
        }

        TopicServicesResponse<ServicesReplyData> serviceReply = abstractWaylineService.inFlightWaylineTaskPrepare(
                SDKManager.getDeviceSDK(waylineJob.getDockSn()), flightTask);
        if (!serviceReply.getData().getResult().isSuccess()) {
            log.info("Prepare task ====> Error code: {}", serviceReply.getData().getResult());
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .workspaceId(waylineJob.getWorkspaceId())
                    .jobId(waylineJob.getJobId())
                    .executeTime(LocalDateTime.now())
                    .status(WaylineJobStatusEnum.FAILED.getVal())
                    .completedTime(LocalDateTime.now())
                    .code(serviceReply.getData().getResult().getCode()).build());
            return false;
        }
        return true;
    }

    /**
     * 取消空中航线
     * @param workspaceId
     * @param dockSn
     */
    @Override
    public void cancelInFlightTask(String workspaceId, String dockSn) {

        // 是否需要判断 无人机的状态  工作中  航线？

        this.publishCancelInFlightTask(workspaceId, dockSn);

    }

    @Override
    public void publishCancelInFlightTask(String workspaceId, String dockSn) {
        boolean isOnline = deviceRedisService.checkDeviceOnline(dockSn);
        if (!isOnline) {
            throw new RuntimeException("Dock is offline.");
        }

        // 取消空中航线
        TopicServicesResponse<ServicesReplyData> serviceReply = abstractWaylineService.inFlightWaylineCancel(SDKManager.getDeviceSDK(dockSn), new BaseModel());
        if (!serviceReply.getData().getResult().isSuccess()) {
            log.info("Cancel job ====> Error: {}", serviceReply.getData().getResult());
            throw new RuntimeException("Failed to cancel the wayline job of " + dockSn);
        }
        // 还需要查询 对应的空中下发航线的in_flight_id;

//        for (String jobId : jobIds) {
//            waylineJobService.updateJob(WaylineJobDTO.builder()
//                    .workspaceId(workspaceId)
//                    .jobId(jobId)
//                    .status(WaylineJobStatusEnum.CANCEL.getVal())
//                    .completedTime(LocalDateTime.now())
//                    .build());
//            RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, workspaceId + RedisConst.DELIMITER + dockSn + RedisConst.DELIMITER + jobId);
//        }

    }

}
