package com.ard.tianyu.service.impl;

import com.ard.tianyu.model.receiver.BoxStreamUrlReceiver;
import com.ard.tianyu.service.ITyService;
import com.ard.tianyu.websocket.utils.WebSocketUtil;
import com.ard.vtdu.domain.ArdVtdu;
import com.ard.vtdu.service.IMediaService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@Service
@Slf4j
public class UavStreamManager {

    private static final String STREAM_LOG_PREFIX = "【stream】";
    private static final String STREAM_NOT_FOUND_MSG = STREAM_LOG_PREFIX + "Box stream not found: ";
    private static final String STREAM_CLOSED_MSG = STREAM_LOG_PREFIX + "Stream closed due to no active users: ";
    private static final String STREAM_IN_USE_MSG = STREAM_LOG_PREFIX + "Stream is still in use, remaining users: ";

    @Resource
    private ITyService tyService;
    @Resource
    private IMediaService mediaService;

    // 盒子拉流地址缓存 key:boxSn value: webrtcUrl
    public static final ConcurrentMap<String, String> BOX_STREAM_PULL_MAP = new ConcurrentHashMap<>();
    // 盒子拉流用户集合缓存 key:boxSn value:Set<UserId>
    public static final ConcurrentMap<String, Set<String>> STREAM_USER_MAP = new ConcurrentHashMap<>();

    /**
     * 开始拉流
     *
     * @param boxSn  盒子序列号
     * @param userId 用户ID
     */
    public void startPullStream(String boxSn, String userId) {
        STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);

        String webrtcUrl = BOX_STREAM_PULL_MAP.get(boxSn);
        if (webrtcUrl != null) {
            log.info("{}Already pulled stream: {}", STREAM_LOG_PREFIX, webrtcUrl);
            sendMessageToUser(userId, webrtcUrl);
            return;
        }

        try {
            BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
            if (boxLiveUrl == null) {
                log.warn("{}{}{}", STREAM_LOG_PREFIX, STREAM_NOT_FOUND_MSG, boxSn);
                sendMessageToUser(userId, STREAM_NOT_FOUND_MSG + boxSn);
                return;
            }

            ArdVtdu vtdu = ArdVtdu.builder()
                    .name(boxSn)
                    .mode(0)
                    .isCode(0)
                    .rtspSource(boxLiveUrl.getRtsp())
                    .build();
            ArdVtdu ardVtdu = mediaService.addPath(vtdu);
            webrtcUrl = ardVtdu.getWebrtcUrl();
            BOX_STREAM_PULL_MAP.put(boxSn, webrtcUrl);

            log.info("{}First pulled stream: {}", STREAM_LOG_PREFIX, webrtcUrl);
            sendMessageToUser(userId, webrtcUrl);
        } catch (Exception e) {
            log.error("{}Failed to start pull stream for box: {}", STREAM_LOG_PREFIX, boxSn, e);
            sendMessageToUser(userId, "Failed to start pull stream due to an internal error.");
        }
    }

    /**
     * 停止拉流
     *
     * @param boxSn  盒子序列号
     * @param userId 用户ID
     */
    public void stopPullStream(String boxSn, String userId) {
        synchronized (this) {
            Set<String> users = STREAM_USER_MAP.get(boxSn);
            if (users == null) {
                log.warn("{}No active users for box: {}", STREAM_LOG_PREFIX, boxSn);
                sendMessageToUser(userId, "No active users for box: " + boxSn);
                return;
            }

            users.remove(userId);
            if (users.isEmpty()) {
                mediaService.removePath(boxSn);
                BOX_STREAM_PULL_MAP.remove(boxSn);
                STREAM_USER_MAP.remove(boxSn);
                log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_CLOSED_MSG, boxSn);
                sendMessageToUser(userId, STREAM_CLOSED_MSG + boxSn);
            } else {
                log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_IN_USE_MSG, users);
                sendMessageToUser(userId, STREAM_IN_USE_MSG + users);
            }
        }
    }
    /**
     * 停止拉流
     *
     * @param userId 用户ID
     */
    public void stopAllPullStreamByUser(String userId) {
        synchronized (this) {
            STREAM_USER_MAP.forEach((boxSn, users) -> stopPullStream(boxSn, userId));
        }
    }

    /**
     * 发送消息给指定用户
     *
     * @param userId  用户ID
     * @param message 消息内容
     */
    private void sendMessageToUser(String userId, String message) {
        try {
            WebSocketUtil.sendMessage(userId, message);
        } catch (Exception e) {
            log.error("{}Failed to send message to user: {}", STREAM_LOG_PREFIX, userId, e);
        }
    }
}