package com.cmvr.framework.websocket.service;

import com.cmvr.common.exception.GlobalException;
import com.cmvr.framework.websocket.manager.ChannelSubscriptionManager;
import com.cmvr.framework.websocket.message.WSChannelSubscriptionMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.BinaryMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.zip.GZIPOutputStream;


@Service
public class MessagePushService {

    private static final Logger log = LoggerFactory.getLogger(MessagePushService.class);

    @Autowired
    private ChannelSubscriptionManager subscriptionManager;

    private final ObjectMapper objectMapper = new ObjectMapper();

    public void pushToChannel(String channel, Object payload) {
        try {
            WSChannelSubscriptionMessage message = new WSChannelSubscriptionMessage();
            message.setChannel(channel);
            message.setTimestamp(System.currentTimeMillis());
            message.setPayload(payload);

            String json = objectMapper.writeValueAsString(message);

            byte[] data = json.getBytes();

            // 压缩数据
            byte[] compressedData = compress(data);

            BinaryMessage binaryMessage = new BinaryMessage(compressedData);
            if (!subscriptionManager.hasSubscribers(channel)) {
                throw new GlobalException("没有订阅该频道的会话");
            }
            subscriptionManager.getSubscribedSessions(channel).stream()
                    .filter(WebSocketSession::isOpen)
                    .forEach(session -> {
                        try {
                            session.sendMessage(binaryMessage);
                        } catch (IOException e) {
                            log.error("消息发送失败: {}", e.getMessage());
                        }
                    });

        } catch (GlobalException e) {
            throw new GlobalException(e.getMessage());
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    private byte[] compress(byte[] data) throws IOException {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        try (GZIPOutputStream gzip = new GZIPOutputStream(baos)) {
            gzip.write(data);
        }
        return baos.toByteArray();
    }
}
