package cn.iocoder.yudao.module.iot.service.rule.data.action.websocket;

import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkWebSocketConfig;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * IoT WebSocket 客户端
 * <p>
 * 负责与外部 WebSocket 服务器建立连接并发送设备消息
 * 支持 ws:// 和 wss:// 协议，支持 JSON 和 TEXT 数据格式
 * 基于 OkHttp WebSocket 实现，兼容 JDK 8+
 * <p>
 * 注意：该类的线程安全由调用方（IotWebSocketDataRuleAction）通过分布式锁保证
 *
 * @author HUIHUI
 */
@Slf4j
public class IotWebSocketClient {

    private final String serverUrl;
    private final Integer connectTimeoutMs;
    private final Integer sendTimeoutMs;
    private final String dataFormat;

    private OkHttpClient okHttpClient;
    private volatile WebSocket webSocket;
    private final AtomicBoolean connected = new AtomicBoolean(false);

    /**
     * WebSocket 正常关闭状态码
     *
     * @see <a href="https://tools.ietf.org/html/rfc6455#section-7.4.1">RFC 6455 - 定义的状态码</a>
     */
    private static final int NORMAL_CLOSURE_STATUS = 1000;

    public IotWebSocketClient(String serverUrl, Integer connectTimeoutMs, Integer sendTimeoutMs, String dataFormat) {
        this.serverUrl = serverUrl;
        this.connectTimeoutMs = connectTimeoutMs != null ? connectTimeoutMs : IotDataSinkWebSocketConfig.DEFAULT_CONNECT_TIMEOUT_MS;
        this.sendTimeoutMs = sendTimeoutMs != null ? sendTimeoutMs : IotDataSinkWebSocketConfig.DEFAULT_SEND_TIMEOUT_MS;
        this.dataFormat = dataFormat != null ? dataFormat : IotDataSinkWebSocketConfig.DEFAULT_DATA_FORMAT;
    }

    /**
     * 连接到 WebSocket 服务器
     * <p>
     * 注意：调用方需要通过分布式锁保证并发安全
     */
    public void connect() throws Exception {
        if (connected.get()) {
            log.warn("[connect][WebSocket 客户端已经连接，无需重复连接]");
            return;
        }

        try {
            // 创建 OkHttpClient
            okHttpClient = new OkHttpClient.Builder()
                    .connectTimeout(connectTimeoutMs, TimeUnit.MILLISECONDS)
                    .readTimeout(sendTimeoutMs, TimeUnit.MILLISECONDS)
                    .writeTimeout(sendTimeoutMs, TimeUnit.MILLISECONDS)
                    .build();

            // 创建 WebSocket 请求
            Request request = new Request.Builder()
                    .url(serverUrl)
                    .build();

            // 使用 CountDownLatch 等待连接完成
            CountDownLatch connectLatch = new CountDownLatch(1);
            AtomicBoolean connectSuccess = new AtomicBoolean(false);
            // 创建 WebSocket 连接
            webSocket = okHttpClient.newWebSocket(request, new IotWebSocketListener(connectLatch, connectSuccess));

            // 等待连接完成
            boolean await = connectLatch.await(connectTimeoutMs, TimeUnit.MILLISECONDS);
            if (!await || !connectSuccess.get()) {
                close();
                throw new Exception("WebSocket 连接超时或失败，服务器地址: " + serverUrl);
            }
            log.info("[connect][WebSocket 客户端连接成功，服务器地址: {}]", serverUrl);
        } catch (Exception e) {
            close();
            log.error("[connect][WebSocket 客户端连接失败，服务器地址: {}]", serverUrl, e);
            throw e;
        }
    }

    /**
     * 发送设备消息
     *
     * @param message 设备消息
     * @throws Exception 发送异常
     */
    public void sendMessage(IotDeviceMessage message) throws Exception {
        WebSocket ws = this.webSocket;
        if (!connected.get() || ws == null) {
            throw new IllegalStateException("WebSocket 客户端未连接");
        }

        try {
            String messageData;
            if (IotDataSinkWebSocketConfig.DEFAULT_DATA_FORMAT.equalsIgnoreCase(dataFormat)) {
                messageData = JsonUtils.toJsonString(message);
            } else {
                messageData = message.toString();
            }

            // 发送消息
            boolean success = ws.send(messageData);
            if (!success) {
                throw new Exception("WebSocket 发送消息失败，消息队列已满或连接已关闭");
            }
            log.debug("[sendMessage][发送消息成功，设备 ID: {}，消息长度: {}]",
                    message.getDeviceId(), messageData.length());
        } catch (Exception e) {
            log.error("[sendMessage][发送消息失败，设备 ID: {}]", message.getDeviceId(), e);
            throw e;
        }
    }

    /**
     * 关闭连接
     */
    public void close() {
        try {
            if (webSocket != null) {
                // 发送正常关闭帧
                webSocket.close(NORMAL_CLOSURE_STATUS, "客户端主动关闭");
                webSocket = null;
            }
            if (okHttpClient != null) {
                // 关闭连接池和调度器
                okHttpClient.dispatcher().executorService().shutdown();
                okHttpClient.connectionPool().evictAll();
                okHttpClient = null;
            }
            connected.set(false);
            log.info("[close][WebSocket 客户端连接已关闭，服务器地址: {}]", serverUrl);
        } catch (Exception e) {
            log.error("[close][关闭 WebSocket 客户端连接异常]", e);
        }
    }

    /**
     * 检查连接状态
     *
     * @return 是否已连接
     */
    public boolean isConnected() {
        return connected.get() && webSocket != null;
    }

    @Override
    public String toString() {
        return "IotWebSocketClient{" +
                "serverUrl='" + serverUrl + '\'' +
                ", dataFormat='" + dataFormat + '\'' +
                ", connected=" + connected.get() +
                '}';
    }

    /**
     * OkHttp WebSocket 监听器
     */
    @SuppressWarnings("NullableProblems")
    private class IotWebSocketListener extends WebSocketListener {

        private final CountDownLatch connectLatch;
        private final AtomicBoolean connectSuccess;

        public IotWebSocketListener(CountDownLatch connectLatch, AtomicBoolean connectSuccess) {
            this.connectLatch = connectLatch;
            this.connectSuccess = connectSuccess;
        }

        @Override
        public void onOpen(WebSocket webSocket, Response response) {
            connected.set(true);
            connectSuccess.set(true);
            connectLatch.countDown();
            log.info("[onOpen][WebSocket 连接已打开，服务器: {}]", serverUrl);
        }

        @Override
        public void onMessage(WebSocket webSocket, String text) {
            log.debug("[onMessage][收到消息: {}]", text);
        }

        @Override
        public void onClosing(WebSocket webSocket, int code, String reason) {
            connected.set(false);
            log.info("[onClosing][WebSocket 正在关闭，code: {}, reason: {}]", code, reason);
        }

        @Override
        public void onClosed(WebSocket webSocket, int code, String reason) {
            connected.set(false);
            log.info("[onClosed][WebSocket 已关闭，code: {}, reason: {}]", code, reason);
        }

        @Override
        public void onFailure(WebSocket webSocket, Throwable t, Response response) {
            connected.set(false);
            connectLatch.countDown(); // 确保连接失败时也释放等待
            log.error("[onFailure][WebSocket 连接失败]", t);
        }
    }

}