package com.webchat.service.listener;

import com.webchat.common.enums.RoleCodeEnum;
import com.webchat.common.util.JsonUtil;
import com.webchat.common.util.SpringContextUtil;
import com.webchat.controller.client.ChatWebSocket;
import com.webchat.domain.vo.request.mess.ChatMessageRequestVO;
import com.webchat.domain.vo.response.UserBaseResponseInfoVO;
import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
import com.webchat.service.UserService;
import com.webchat.service.ai.AiAgentFlowService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.websocket.Session;
import java.io.IOException;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;

@Slf4j
@Component
public class RedisChatMessageListener implements MessageListener {

    private final RedisTemplate redisTemplate;

    private static final int QUEUE_SIZE = 5000;

    private static ArrayBlockingQueue<String> messageQueue = new ArrayBlockingQueue<>(QUEUE_SIZE);

    public RedisChatMessageListener(@Qualifier("redisTemplate") RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void onMessage(Message message, byte[] bytes) {
        String channel = (String) redisTemplate.getStringSerializer().deserialize(message.getChannel());
        String messageStr = (String) redisTemplate.getValueSerializer().deserialize(message.getBody());
        log.info("redis message listener ======> channel:{}, message:{}", channel, messageStr);
        try {
            messageQueue.put(messageStr);
        } catch (InterruptedException e) {
            log.error("消息加入队列异常", e);
        }
    }

    /**
     * 队列消费，此处使用单独线程处理队列消息
     *
     */
    public void consumeMessageQueue() {
        while (true) {
            try {
                String message = messageQueue.take();
                if (StringUtils.isNotBlank(message)) {
                    this.handleChatMessage(message);
                }
            } catch (InterruptedException e) {
                log.error("队列消息消费异常", e);
            }
        }
    }

    private void handleChatMessage(String message) {
        try {
            log.info("来自客户端消息 =====> message:{}", message);
            UserService userService = SpringContextUtil.getBean(UserService.class);
            ChatMessageRequestVO chatMessage = JsonUtil.fromJson(message, ChatMessageRequestVO.class);
            // 设置消息发送人信息、接收人信息（支持群聊模式）
            String sendUserId = chatMessage.getSenderId();
            String receiverId = chatMessage.getReceiverId();
            UserBaseResponseInfoVO receiver = userService.getUserInfoByUserId(receiverId);
            if (RoleCodeEnum.ROBOT.getCode().equals(receiver.getRoleCode())) {
                /**
                 * 机器人对话，走模型处理，处理结果使用SSE推给客户端
                 */
                AiAgentFlowService aiAgentFlowService = SpringContextUtil.getBean(AiAgentFlowService.class);
                aiAgentFlowService.doAgent(chatMessage);
                return;
            }

            /**
             * websocket 消息推送
             */
            ChatMessageResponseVO chatMessageResponse = new ChatMessageResponseVO();
            chatMessageResponse.setMessage(chatMessage.getMessage());
            chatMessageResponse.setSenderId(chatMessage.getSenderId());
            chatMessageResponse.setReceiverId(chatMessage.getReceiverId());
            chatMessageResponse.setTime(new Date().getTime());
            UserBaseResponseInfoVO sender = userService.getUserInfoByUserId(sendUserId);
            userService.getUserInfoByUserId(sendUserId);
            chatMessageResponse.setSender(sender);
            chatMessageResponse.setReceiverId(receiverId);
            chatMessageResponse.setReceiver(receiver);
            chatMessageResponse.setRejectVideoOffer(chatMessage.getRejectVideoOffer());
            chatMessageResponse.setType(chatMessage.getType());
            // 校验消息接收人是否为群组
            boolean isGroup = RoleCodeEnum.GROUP.getCode().equals(receiver.getRoleCode());
            chatMessageResponse.setGroupMessage(isGroup);
            Set<String> receiverUserIds = new HashSet<>();
            if (isGroup) {
                // 特殊处理来自群组的消息
                // 这里我们引入一个代理消息发送人proxyUser: 做一个角色转换，将群组作为消息发送人，代理原消息发送人
                chatMessageResponse.setSenderId(receiverId); // 群组
                chatMessageResponse.setSender(receiver);
                // 被代理的消息发送人
                chatMessageResponse.setProxySenderId(sendUserId); // 代理实际消息发送人
                chatMessageResponse.setProxySender(sender);
                // 获取群聊下所有用户，走redis缓存查询群组下的所有用户
                List<String> groupUserIds = userService.getGroupUserIdsFromCache(receiverId);
                if (CollectionUtils.isNotEmpty(groupUserIds)) {
                    receiverUserIds.addAll(groupUserIds);
                }
            } else {
                receiverUserIds.add(receiverId);
            }
            // 删除消息发送人（不支持自己发送给自己）
            receiverUserIds.remove(sendUserId);
            sendMessageTo(JsonUtil.toJsonString(chatMessageResponse), receiverUserIds);
        } catch (Exception e) {
            log.info("onMessage error. message:{}", message);
        }
    }

    public void sendMessageTo(String message, Set<String> receivers) throws IOException {
        if (CollectionUtils.isEmpty(receivers)) {
            return;
        }
        for (ChatWebSocket chat : ChatWebSocket.clients.values()) {
            // 支持点对点，群聊模式
            if (receivers.contains(chat.getUserId())) {
                Session session = chat.getSession();
                if (session != null && session.isOpen()) {
                    session.getAsyncRemote().sendText(message);
                    log.info("websocket消息发送成功，接收人:{} 消息内容:{}", chat.getUserId(), JsonUtil.toJsonString(message));
                }
            }
        }
    }


}
