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

import com.dji.sample.component.redis.RedisConst;
import com.dji.sample.component.redis.RedisOpsUtils;
import com.dji.sample.manage.model.dto.*;
import com.dji.sample.manage.model.param.DeviceQueryParam;
import com.dji.sample.manage.service.*;
import com.dji.sdk.cloudapi.device.DeviceDomainEnum;
import com.dji.sdk.cloudapi.device.VideoId;
import com.dji.sdk.cloudapi.livestream.*;
import com.dji.sdk.cloudapi.livestream.api.AbstractLivestreamService;
import com.dji.sdk.common.HttpResultResponse;
import com.dji.sdk.common.SDKManager;
import com.dji.sdk.mqtt.services.ServicesReplyData;
import com.dji.sdk.mqtt.services.TopicServicesResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * @author sean.zhou
 * @date 2021/11/22
 * @version 0.1
 */
@Service
@Transactional
@Slf4j
public class LiveStreamServiceImpl implements ILiveStreamService {

    private static final int LIVE_ALREADY_STREAMING_CODE = 513003;

    @Autowired
    private ICapacityCameraService capacityCameraService;

    @Autowired
    private IDeviceService deviceService;

    @Autowired
    private IWorkspaceService workspaceService;

    @Autowired
    private IDeviceRedisService deviceRedisService;

    @Autowired
    private AbstractLivestreamService abstractLivestreamService;

    @Override
    public CapacityDeviceDTO getLiveCapacity(String workspaceId, String deviceSn) {

        // Query the devices in this workspace.
        List<DeviceDTO> devicesList = deviceService.getDevicesByParams(
                DeviceQueryParam.builder()
                        .workspaceId(workspaceId)
                        .domains(List.of(DeviceDomainEnum.DRONE.getDomain(), DeviceDomainEnum.DOCK.getDomain()))
                        .deviceSn(deviceSn)
                        .build());

        CapacityDeviceDTO capacityDeviceDTO = new CapacityDeviceDTO();
        if (CollectionUtils.isEmpty(devicesList)) {
            return capacityDeviceDTO;
        }
        DeviceDTO device = devicesList.get(0);
        capacityDeviceDTO.setName(Objects.requireNonNullElse(device.getNickname(), device.getDeviceName()));
        capacityDeviceDTO.setSn(device.getDeviceSn());
        capacityDeviceDTO.setCamerasList(capacityCameraService.getCapacityCameraByDeviceSn(device.getDeviceSn()));

        return capacityDeviceDTO;
    }

    @Override
    public CapacityDeviceDTO getLiveCapacity2(String deviceSn) {

        // Query the devices in this workspace.
        List<DeviceDTO> devicesList = deviceService.getDevicesByParams(
                DeviceQueryParam.builder()
                        .domains(List.of(DeviceDomainEnum.DRONE.getDomain(), DeviceDomainEnum.DOCK.getDomain()))
                        .deviceSn(deviceSn)
                        .build());

        CapacityDeviceDTO capacityDeviceDTO = new CapacityDeviceDTO();
        if (CollectionUtils.isEmpty(devicesList)) {
            return capacityDeviceDTO;
        }
        DeviceDTO device = devicesList.get(0);
        capacityDeviceDTO.setName(Objects.requireNonNullElse(device.getNickname(), device.getDeviceName()));
        capacityDeviceDTO.setSn(device.getDeviceSn());
        capacityDeviceDTO.setCamerasList(capacityCameraService.getCapacityCameraByDeviceSn(device.getDeviceSn()));

        return capacityDeviceDTO;
    }

    @Override
    public List<CapacityDeviceDTO> getLiveCapacity(String workspaceId) {

        // Query all devices in this workspace.
        List<DeviceDTO> devicesList = deviceService.getDevicesByParams(
                DeviceQueryParam.builder()
                        .workspaceId(workspaceId)
                        .domains(List.of(DeviceDomainEnum.DRONE.getDomain(), DeviceDomainEnum.DOCK.getDomain()))
                        .build());

        // Query the live capability of each drone.
        return devicesList.stream()
                .filter(device -> deviceRedisService.checkDeviceOnline(device.getDeviceSn()))
                .map(device -> CapacityDeviceDTO.builder()
                        .name(Objects.requireNonNullElse(device.getNickname(), device.getDeviceName()))
                        .sn(device.getDeviceSn())
                        .camerasList(capacityCameraService.getCapacityCameraByDeviceSn(device.getDeviceSn()))
                        .build())
                .collect(Collectors.toList());
    }

    @Override
    public HttpResultResponse liveStart(LiveTypeDTO liveParam) {
        // Check if this lens is available live.
        HttpResultResponse<DeviceDTO> responseResult = this.checkBeforeLive(liveParam.getVideoId());
        if (HttpResultResponse.CODE_SUCCESS != responseResult.getCode()) {
            return responseResult;
        }


        ILivestreamUrl url = LiveStreamProperty.get(liveParam.getUrlType());
        url = setExt(liveParam.getUrlType(), url, liveParam.getVideoId());

        // ========== 优化：先查 Redis，已有活跃会话则直接复用，跳过 MQTT 推流 ==========
        Set<String> existingSessions = findSessionKeys(liveParam.getVideoId());
        boolean streamAlreadyRunning = existingSessions != null && !existingSessions.isEmpty();
        String rtspOutputUrl = null; // RTSP 特殊：播放地址来自设备响应

        if (streamAlreadyRunning) {
            log.info("Stream already running with {} viewer(s), skip MQTT push and reuse. videoId={}",
                    existingSessions.size(), liveParam.getVideoId());
        } else {
            // 首个观看者，真正向设备发送 MQTT 推流指令
            TopicServicesResponse<ServicesReplyData<String>> response = abstractLivestreamService.liveStartPush(
                    SDKManager.getDeviceSDK(responseResult.getData().getDeviceSn()),
                    new LiveStartPushRequest()
                            .setUrl(url)
                            .setUrlType(liveParam.getUrlType())
                            .setVideoId(liveParam.getVideoId())
                            .setVideoQuality(liveParam.getVideoQuality()));

            if (!response.getData().getResult().isSuccess()) {
                int errorCode = response.getData().getResult().getCode();
                // 设备报"已在直播"但 Redis 无会话（可能 Redis 被清过），视为可复用
                if (errorCode == LIVE_ALREADY_STREAMING_CODE) {
                    log.info("Device reports already streaming but no Redis sessions, treating as reuse. videoId={}",
                            liveParam.getVideoId());
                } else {
                    return HttpResultResponse.error(response.getData().getResult());
                }
            }

            // 保存 RTSP 设备返回的 output（仅首次推流时设备会返回）
            if (StringUtils.hasText(response.getData().getOutput())) {
                rtspOutputUrl = response.getData().getOutput();
            }
        }

        // ========== 构造播放地址（无论首个还是复用，URL 构造逻辑一致） ==========
        LiveDTO live = new LiveDTO();

        switch (liveParam.getUrlType()) {
            case AGORA:
                break;
            case RTMP:
                live.setUrl(url.toString().replace("rtmp", "webrtc"));
                break;
            case GB28181:
                LivestreamGb28181Url gb28181 = (LivestreamGb28181Url) url;
                live.setUrl(new StringBuilder()
                        .append("webrtc://")
                        .append(gb28181.getServerIP())
                        .append("/live/")
                        .append(gb28181.getAgentID())
                        .append("@")
                        .append(gb28181.getChannel())
                        .toString());
                break;
            case RTSP:
                // RTSP 优先用设备返回的 output 地址，复用时回退到配置 URL
                if (StringUtils.hasText(rtspOutputUrl)) {
                    live.setUrl(rtspOutputUrl);
                } else {
                    live.setUrl(url.toString());
                }
                break;
            case WHIP:
                live.setUrl(url.toString().replace("whip", "whep"));
                break;
            default:
                return HttpResultResponse.error(LiveErrorCodeEnum.URL_TYPE_NOT_SUPPORTED);
        }

        // ========== 无论首个还是复用，都注册新的观看者会话 ==========
        // 如果前端传了 clientId，先清理同一 clientId 的旧会话（处理页面刷新场景）
        if (StringUtils.hasText(liveParam.getClientId())) {
            cleanupClientSessions(liveParam.getVideoId(), liveParam.getClientId());
        }

        String sessionId = UUID.randomUUID().toString();
        registerSession(liveParam.getVideoId(), sessionId, liveParam.getClientId());
        live.setSessionId(sessionId);
        log.info("Registered live session sessionId={}, clientId={}, videoId={}, totalViewers={}",
                sessionId, liveParam.getClientId(), liveParam.getVideoId(),
                countSessionKeys(liveParam.getVideoId()));

        return HttpResultResponse.success(live);
    }

    @Override
    public HttpResultResponse liveStop(VideoId videoId) {
        HttpResultResponse<DeviceDTO> responseResult = this.checkBeforeLive(videoId);
        if (HttpResultResponse.CODE_SUCCESS != responseResult.getCode()) {
            return responseResult;
        }

        TopicServicesResponse<ServicesReplyData> response = abstractLivestreamService.liveStopPush(
                SDKManager.getDeviceSDK(responseResult.getData().getDeviceSn()), new LiveStopPushRequest()
                        .setVideoId(videoId));
        if (!response.getData().getResult().isSuccess()) {
            return HttpResultResponse.error(response.getData().getResult());
        }

        return HttpResultResponse.success();
    }

    @Override
    public HttpResultResponse liveSetQuality(LiveTypeDTO liveParam) {
        HttpResultResponse<DeviceDTO> responseResult = this.checkBeforeLive(liveParam.getVideoId());
        if (responseResult.getCode() != 0) {
            return responseResult;
        }

        TopicServicesResponse<ServicesReplyData> response = abstractLivestreamService.liveSetQuality(
                SDKManager.getDeviceSDK(responseResult.getData().getDeviceSn()), new LiveSetQualityRequest()
                        .setVideoQuality(liveParam.getVideoQuality())
                        .setVideoId(liveParam.getVideoId()));
        if (!response.getData().getResult().isSuccess()) {
            return HttpResultResponse.error(response.getData().getResult());
        }

        return HttpResultResponse.success();
    }

    @Override
    public HttpResultResponse liveLensChange(LiveTypeDTO liveParam) {
        HttpResultResponse<DeviceDTO> responseResult = this.checkBeforeLive(liveParam.getVideoId());
        if (HttpResultResponse.CODE_SUCCESS != responseResult.getCode()) {
            return responseResult;
        }

        TopicServicesResponse<ServicesReplyData> response = abstractLivestreamService.liveLensChange(
                SDKManager.getDeviceSDK(responseResult.getData().getDeviceSn()), new LiveLensChangeRequest()
                        .setVideoType(liveParam.getVideoType())
                        .setVideoId(liveParam.getVideoId()));

        if (!response.getData().getResult().isSuccess()) {
            return HttpResultResponse.error(response.getData().getResult());
        }

        return HttpResultResponse.success();
    }

    @Override
    public HttpResultResponse liveCameraChange(LiveTypeDTO liveParam) {
        HttpResultResponse<DeviceDTO> responseResult = this.checkBeforeLive(liveParam.getVideoId());
        if (HttpResultResponse.CODE_SUCCESS != responseResult.getCode()) {
            return responseResult;
        }
        TopicServicesResponse<ServicesReplyData> response = abstractLivestreamService.liveCameraChange(
                SDKManager.getDeviceSDK(responseResult.getData().getDeviceSn()), new LiveCameraChangeRequest()
                        .setCameraPosition(liveParam.getCameraPosition())
                        .setVideoId(liveParam.getVideoId()));

        if (!response.getData().getResult().isSuccess()) {
            return HttpResultResponse.error(response.getData().getResult());
        }

        return HttpResultResponse.success();
    }

    /**
     * Check if this lens is available live.
     * @param videoId
     * @return
     */
    private HttpResultResponse<DeviceDTO> checkBeforeLive(VideoId videoId) {
        if (Objects.isNull(videoId)) {
            return HttpResultResponse.error(LiveErrorCodeEnum.ERROR_PARAMETERS);
        }

        Optional<DeviceDTO> deviceOpt = deviceService.getDeviceBySn(videoId.getDroneSn());
        // Check if the gateway device connected to this drone exists
        if (deviceOpt.isEmpty()) {
            return HttpResultResponse.error(LiveErrorCodeEnum.NO_AIRCRAFT);
        }

        if (DeviceDomainEnum.DOCK == deviceOpt.get().getDomain()) {
            return HttpResultResponse.success(deviceOpt.get());
        }
        List<DeviceDTO> gatewayList = deviceService.getDevicesByParams(
                DeviceQueryParam.builder()
                        .childSn(videoId.getDroneSn())
                        .build());
        if (gatewayList.isEmpty()) {
            return HttpResultResponse.error(LiveErrorCodeEnum.NO_FLIGHT_CONTROL);
        }

        return HttpResultResponse.success(gatewayList.get(0));
    }

    /**
     * This is business-customized logic and is only used for testing.
     * @param type
     * @param url
     * @param videoId
     */
    private ILivestreamUrl setExt(UrlTypeEnum type, ILivestreamUrl url, VideoId videoId) {
        switch (type) {
            case AGORA:
                LivestreamAgoraUrl agoraUrl = (LivestreamAgoraUrl) url.clone();
                return agoraUrl.setSn(videoId.getDroneSn());
            case RTMP:
                LivestreamRtmpUrl rtmpUrl = (LivestreamRtmpUrl) url.clone();
                return rtmpUrl.setUrl(rtmpUrl.getUrl() + videoId.getDroneSn() + "-" + videoId.getPayloadIndex().toString());
            case GB28181:
                String random = String.valueOf(Math.abs(videoId.getDroneSn().hashCode()) % 1000);
                LivestreamGb28181Url gbUrl = (LivestreamGb28181Url) url.clone();
                gbUrl.setAgentID(gbUrl.getAgentID().substring(0, 20 - random.length()) + random);
                String deviceType = String.valueOf(videoId.getPayloadIndex().getType().getType());
                return gbUrl.setChannel(gbUrl.getChannel().substring(0, 20 - deviceType.length()) + deviceType);
            case WHIP:
                LivestreamWhipUrl whipUrl = (LivestreamWhipUrl) url.clone();
                return whipUrl.setUrl(whipUrl.getUrl() + videoId.getDroneSn() + "-" + videoId.getPayloadIndex().toString());
        }
        return url;
    }

    // ==================== 直播会话管理（多用户防冲突 + 心跳保活） ====================

    /**
     * 生成观看者会话 Redis key。
     * 格式: live_session:{droneSn}_{payloadIndex}:{sessionId}
     */
    private String buildSessionKey(VideoId videoId, String sessionId) {
        String videoIdStr = videoId.getDroneSn() + "_" + videoId.getPayloadIndex().toString();
        return RedisConst.LIVE_SESSION_PREFIX + videoIdStr + RedisConst.DELIMITER + sessionId;
    }

    /**
     * 查找某路直播流下所有活跃观看者会话 key。
     * 模式: live_session:{droneSn}_{payloadIndex}:*
     */
    private Set<String> findSessionKeys(VideoId videoId) {
        String pattern = RedisConst.LIVE_SESSION_PREFIX
                + videoId.getDroneSn() + "_" + videoId.getPayloadIndex().toString()
                + RedisConst.DELIMITER + "*";
        return RedisOpsUtils.getAllKeys(pattern);
    }

    /**
     * 注册观看者会话到 Redis，设置 TTL。
     * value 存储 clientId（用于刷新去重），无 clientId 则存 sessionId。
     */
    private void registerSession(VideoId videoId, String sessionId, String clientId) {
        String key = buildSessionKey(videoId, sessionId);
        String value = StringUtils.hasText(clientId) ? clientId : sessionId;
        RedisOpsUtils.setWithExpire(key, value, RedisConst.LIVE_SESSION_TTL_SECOND);
    }

    /**
     * 清理同一 clientId 的旧会话（处理页面刷新 / 重复开流场景）。
     * 扫描该 videoId 下所有会话 key，value == clientId 的全部删除。
     */
    private void cleanupClientSessions(VideoId videoId, String clientId) {
        Set<String> keys = findSessionKeys(videoId);
        if (keys == null || keys.isEmpty()) {
            return;
        }
        for (String key : keys) {
            Object val = RedisOpsUtils.get(key);
            if (clientId.equals(val)) {
                RedisOpsUtils.del(key);
                log.info("Cleaned up stale session for clientId={}, key={}", clientId, key);
            }
        }
    }

    /**
     * 统计某路直播流当前活跃观看者数量。
     */
    private int countSessionKeys(VideoId videoId) {
        Set<String> keys = findSessionKeys(videoId);
        return keys != null ? keys.size() : 0;
    }

    @Override
    public HttpResultResponse liveStopSession(VideoId videoId, String sessionId) {
        if (Objects.isNull(videoId) || !StringUtils.hasText(sessionId)) {
            return HttpResultResponse.error("videoId and sessionId are required");
        }

        // 1. 删除指定 sessionId 的会话 key
        String key = buildSessionKey(videoId, sessionId);
        RedisOpsUtils.del(key);
        log.info("Removed live session sessionId={}, videoId={}", sessionId, videoId);

        // 2. 检查剩余观看者
        Set<String> remaining = findSessionKeys(videoId);
        if (remaining == null || remaining.isEmpty()) {
            // 最后一个观看者离开 → 真正停流
            log.info("No remaining viewers for videoId={}, executing real stop", videoId);
            return this.liveStop(videoId);
        }

        // 还有其他观看者，仅注销当前会话
        log.info("Still {} viewer(s) watching videoId={}, skip real stop", remaining.size(), videoId);
        return HttpResultResponse.success();
    }

    @Override
    public HttpResultResponse liveHeartbeat(VideoId videoId, String sessionId) {
        if (Objects.isNull(videoId) || !StringUtils.hasText(sessionId)) {
            return HttpResultResponse.error("videoId and sessionId are required");
        }

        String key = buildSessionKey(videoId, sessionId);
        if (!RedisOpsUtils.checkExist(key)) {
            return HttpResultResponse.error("Session not found, please restart live");
        }
        RedisOpsUtils.expireKey(key, RedisConst.LIVE_SESSION_TTL_SECOND);
        return HttpResultResponse.success();
    }

    @Override
    public HttpResultResponse liveForceStop(VideoId videoId) {
        if (Objects.isNull(videoId)) {
            return HttpResultResponse.error("videoId is required");
        }

        // 清除所有观看者会话
        Set<String> keys = findSessionKeys(videoId);
        if (keys != null && !keys.isEmpty()) {
            keys.forEach(RedisOpsUtils::del);
            log.info("Force cleared {} session(s) for videoId={}", keys.size(), videoId);
        }

        // 直接执行真正停流
        return this.liveStop(videoId);
    }
}
