package com.dji.sample.component;

import com.dji.sample.component.redis.RedisConst;
import com.dji.sample.component.redis.RedisOpsUtils;
import com.dji.sample.manage.model.dto.DeviceDTO;
import com.dji.sample.manage.service.IDeviceRedisService;
import com.dji.sample.manage.service.IDeviceService;
import com.dji.sdk.cloudapi.device.DeviceDomainEnum;
import com.dji.sdk.common.SDKManager;
import com.dji.sdk.mqtt.IMqttTopicService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Arrays;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * @author sean.zhou
 * @date 2021/11/24
 * @version 0.1
 */
@Component
@Slf4j
public class GlobalScheduleService {

    @Autowired
    private IDeviceService deviceService;

    @Autowired
    private IMqttTopicService topicService;

    @Autowired
    private IDeviceRedisService deviceRedisService;

    @Autowired
    private MqttPahoMessageDrivenChannelAdapter mqttInbound;

    /**
     * Check the status of the devices every 30 seconds. It is recommended to use cache.
     */
    @Scheduled(initialDelay = 10, fixedRate = 30, timeUnit = TimeUnit.SECONDS)
    private void deviceStatusListen() {
        int start = RedisConst.DEVICE_ONLINE_PREFIX.length();

        RedisOpsUtils.getAllKeys(RedisConst.DEVICE_ONLINE_PREFIX + "*").forEach(key -> {
            long expire = RedisOpsUtils.getExpire(key);
            if (expire <= 30) {
                DeviceDTO device = (DeviceDTO) RedisOpsUtils.get(key);
                if (null == device) {
                    return;
                }
                if (DeviceDomainEnum.DRONE == device.getDomain()) {
                    deviceService.subDeviceOffline(key.substring(start));
                } else {
                    deviceService.gatewayOffline(key.substring(start));
                }
                RedisOpsUtils.del(key);
            }
        });

        log.info("Subscriptions: {}", Arrays.toString(topicService.getSubscribedTopic()));
    }

    /**
     * Recover MQTT adapter if down; always re-sync online gateways into SDKManager.
     * Wildcard +/osd can deliver traffic before status re-login after restart.
     */
    @Scheduled(initialDelay = 5, fixedRate = 30, timeUnit = TimeUnit.SECONDS)
    private void mqttConnectionCheck() {
        if (!mqttInbound.isRunning()) {
            log.warn("MQTT adapter is not running, attempting to restart and re-subscribe...");
            try {
                mqttInbound.start();
            } catch (Exception e) {
                log.error("Failed to restart MQTT adapter", e);
                return;
            }
        }
        resubscribeOnlineDevices();
    }

    /**
     * Re-subscribe all currently online non-drone devices and ensure SDKManager registration.
     */
    private void resubscribeOnlineDevices() {
        int start = RedisConst.DEVICE_ONLINE_PREFIX.length();
        RedisOpsUtils.getAllKeys(RedisConst.DEVICE_ONLINE_PREFIX + "*")
                .stream()
                .map(key -> key.substring(start))
                .map(deviceRedisService::getDeviceOnline)
                .filter(Optional::isPresent)
                .map(Optional::get)
                .forEach(device -> {
                    try {
                        if (DeviceDomainEnum.DRONE == device.getDomain()) {
                            deviceService.subDroneOnlineSubscribeTopic(device.getDeviceSn());
                            return;
                        }

                        boolean wasRegistered = SDKManager.isRegistered(device.getDeviceSn());
                        String childSn = device.getChildDeviceSn();
                        String childThingVersion = StringUtils.hasText(childSn)
                                ? deviceRedisService.getDeviceOnline(childSn).map(DeviceDTO::getThingVersion).orElse(null)
                                : null;

                        deviceService.subDeviceOnlineSubscribeTopic(
                                SDKManager.registerDevice(device.getDeviceSn(), childSn,
                                        device.getDomain(), device.getType(), device.getSubType(),
                                        device.getThingVersion(), childThingVersion));
                        if (!wasRegistered) {
                            log.info("Re-registered online gateway into SDKManager: {}", device.getDeviceSn());
                        }
                    } catch (Exception e) {
                        log.error("Failed to re-subscribe device: {}", device.getDeviceSn(), e);
                    }
                });
    }

}
