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

import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkWebSocketConfig;
import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
import cn.iocoder.yudao.module.iot.service.rule.data.action.websocket.IotWebSocketClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * WebSocket 的 {@link IotDataRuleAction} 实现类
 * <p>
 * 负责将设备消息发送到外部 WebSocket 服务器
 * 支持 ws:// 和 wss:// 协议，支持 JSON 和 TEXT 数据格式
 * 使用连接池管理 WebSocket 连接，提高性能和资源利用率
 *
 * @author HUIHUI
 */
@Component
@Slf4j
public class IotWebSocketDataRuleAction extends
        IotDataRuleCacheableAction<IotDataSinkWebSocketConfig, IotWebSocketClient> {

    /**
     * 锁等待超时时间（毫秒）
     */
    private static final long LOCK_WAIT_TIME_MS = 5000;

    /**
     * 重连锁，key 为 WebSocket 服务器地址
     * <p>
     * WebSocket 连接是与特定服务器实例绑定的，使用单机锁即可保证重连的线程安全
     */
    private final ConcurrentHashMap<String, ReentrantLock> reconnectLocks = new ConcurrentHashMap<>();

    @Override
    public Integer getType() {
        return IotDataSinkTypeEnum.WEBSOCKET.getType();
    }

    @Override
    protected IotWebSocketClient initProducer(IotDataSinkWebSocketConfig config) throws Exception {
        // 1. 参数校验
        if (StrUtil.isBlank(config.getServerUrl())) {
            throw new IllegalArgumentException("WebSocket 服务器地址不能为空");
        }
        if (!StrUtil.startWithAny(config.getServerUrl(), "ws://", "wss://")) {
            throw new IllegalArgumentException("WebSocket 服务器地址必须以 ws:// 或 wss:// 开头");
        }

        // 2.1 创建 WebSocket 客户端
        IotWebSocketClient webSocketClient = new IotWebSocketClient(
                config.getServerUrl(),
                config.getConnectTimeoutMs(),
                config.getSendTimeoutMs(),
                config.getDataFormat()
        );
        // 2.2 连接服务器
        webSocketClient.connect();
        log.info("[initProducer][WebSocket 客户端创建并连接成功，服务器: {}，数据格式: {}]",
                config.getServerUrl(), config.getDataFormat());
        return webSocketClient;
    }

    @Override
    protected void closeProducer(IotWebSocketClient producer) throws Exception {
        if (producer != null) {
            producer.close();
        }
    }

    @Override
    protected void execute(IotDeviceMessage message, IotDataSinkWebSocketConfig config) throws Exception {
        try {
            // 1.1 获取或创建 WebSocket 客户端
            IotWebSocketClient webSocketClient = getProducer(config);

            // 1.2 检查连接状态，如果断开则使用分布式锁保证重连的线程安全
            if (!webSocketClient.isConnected()) {
                reconnectWithLock(webSocketClient, config);
            }

            // 2.1 发送消息
            webSocketClient.sendMessage(message);
            // 2.2 记录发送成功日志
            log.info("[execute][message({}) config({}) 发送成功，WebSocket 服务器: {}]",
                    message, config, config.getServerUrl());
        } catch (Exception e) {
            log.error("[execute][message({}) config({}) 发送失败，WebSocket 服务器: {}]",
                    message, config, config.getServerUrl(), e);
            throw e;
        }
    }

    /**
     * 使用锁进行重连，保证同一服务器地址的重连操作线程安全
     *
     * 加锁原因：多线程并发发送消息时，多个线程可能同时检测到连接断开并尝试重连，使用锁 + 双重检查保证同一服务器地址只有一个线程执行重连操作
     *
     * @param webSocketClient WebSocket 客户端
     * @param config          配置信息
     */
    private void reconnectWithLock(IotWebSocketClient webSocketClient, IotDataSinkWebSocketConfig config) throws Exception {
        ReentrantLock lock = reconnectLocks.computeIfAbsent(config.getServerUrl(), k -> new ReentrantLock());
        boolean acquired = false;
        try {
            acquired = lock.tryLock(LOCK_WAIT_TIME_MS, TimeUnit.MILLISECONDS);
            if (!acquired) {
                throw new RuntimeException("获取 WebSocket 重连锁超时，服务器: " + config.getServerUrl());
            }
            // 双重检查：获取锁后再次检查连接状态，避免重复连接
            if (!webSocketClient.isConnected()) {
                log.warn("[reconnectWithLock][WebSocket 连接已断开，尝试重新连接，服务器: {}]", config.getServerUrl());
                webSocketClient.connect();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("获取 WebSocket 重连锁被中断，服务器: " + config.getServerUrl(), e);
        } finally {
            if (acquired && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

}