package com.agri.framework.manager;

import com.agri.common.core.domain.entity.SysUser;
import com.agri.common.utils.SecurityUtils;
import com.agri.system.domain.SysAgriInfo;
import com.agri.system.service.ISysAgriInfoService;
import com.agri.system.service.ISysUserService;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * MQTT订阅关系管理器
 * 核心功能：
 * 1. 维护前端-设备双向订阅关系（Redis）
 * 2. 前端单设备/全量设备订阅/取消订阅
 * 3. Redis批量操作（pipeline），提升性能
 * 4. 清理无效订阅关系，避免脏数据
 * 适配JDK 8，无心跳包相关逻辑
 */
@Component
public class MqttSubscriptionManager {
    /**
     * 优化：统一使用SLF4J日志（JDK 8兼容）
     */
    private static final Logger log = LoggerFactory.getLogger(MqttSubscriptionManager.class);

    /**
     * Redis模板，用于存储订阅关系、设备在线状态、分布式锁
     */
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private ISysUserService sysUserService;

    /**
     * 农业信息服务，查询用户名下设备
     */
    @Resource
    private ISysAgriInfoService agriInfoService;

    // 新增：前端订阅关系TTL（兜底“取消订阅失败/异常退出”）——只维护subc:{clientId}的TTL
    @Value("${spring.mqtt.subc-ttl-seconds:3600}")
    private int subcTtlSeconds;

    // 新增：pipeline 批量 SISMEMBER subc:{clientId} deviceId（N次->1次往返） 拿取失效的client
    public List<Boolean> pipeIsMemberSubc(List<String> clientIds, String deviceId) {
        if (clientIds == null || clientIds.isEmpty() || !StringUtils.hasText(deviceId)) {
            return Collections.emptyList();
        }

        // ✅ 关键：不发“占位命令”；只对有效clientId发SISMEMBER，同时保证返回结果与入参严格对齐
        int n = clientIds.size();
        // 创建长度n全部false的队列
        List<Boolean> out = new ArrayList<>(Collections.nCopies(n, Boolean.FALSE));
        List<Integer> idx = new ArrayList<>(n);

        for (int i = 0; i < n; i++) {
            if (StringUtils.hasText(clientIds.get(i))) {
                // 符合条件存进clientIds对应的索引
                idx.add(i);
            }
        }
        if (idx.isEmpty()) {
            return out;
        }

        // 处理每个每个clientId的值是否还在  值存在rs中  执行 stringRedisTemplate.executePipelined
        List<Object> rs = stringRedisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            RedisSerializer<String> serializer = stringRedisTemplate.getStringSerializer();
            byte[] member = serializer.serialize(deviceId);
            for (int i : idx) {
                String clientId = clientIds.get(i);
                // 存放命令
                connection.sIsMember(serializer.serialize("subc:" + clientId), member);
            }
            return null;
        });

        for (int j = 0; j < idx.size() && j < (rs == null ? 0 : rs.size()); j++) {
            out.set(idx.get(j), Boolean.TRUE.equals(rs.get(j)));
        }
        return out;
    }

    // 对应的subc不存在 删除对应的sub  pipeline 批量 SREM sub:{deviceId} clientId（清理残留 N次->1次往返）
    public void pipeSRemSub(String deviceId, List<String> staleClientIds) {
        if (!StringUtils.hasText(deviceId) || staleClientIds == null || staleClientIds.isEmpty()) {
            return;
        }
        stringRedisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            RedisSerializer<String> serializer = stringRedisTemplate.getStringSerializer();
            byte[] subKey = serializer.serialize("sub:" + deviceId);
            for (String clientId : staleClientIds) {
                if (StringUtils.hasText(clientId)) {
                    connection.sRem(subKey, serializer.serialize(clientId));
                }
            }
            return null;
        });
    }

    /**
     * 前端订阅设备（Controller调用）
     */
    public void subscribeDevice(String clientId, String deviceId) {
        if (!StringUtils.hasText(clientId) || !StringUtils.hasText(deviceId)) {
            log.error("【订阅管理】clientId或deviceId不能为空");
            throw new IllegalArgumentException("clientId和deviceId不能为空");
        }

        // 保存订阅关系到Redis
        stringRedisTemplate.opsForSet().add("sub:" + deviceId, clientId);
        stringRedisTemplate.opsForSet().add("subc:" + clientId, deviceId);

        // 新增：订阅成功后给subc设置TTL（兜底“取消订阅失败/异常退出”）
        stringRedisTemplate.expire("subc:" + clientId, subcTtlSeconds, TimeUnit.SECONDS);

        log.info("【订阅管理】前端{}订阅设备{}成功", clientId, deviceId);
    }

    /**
     * 前端取消订阅设备状态接口（供Controller层调用）
     * 逻辑：从设备的订阅列表移除前端clientId
     *
     * @param clientId 前端唯一标识
     * @param deviceId 设备ID
     */
    public void unsubscribeDevice(String clientId, String deviceId) {
        if (!StringUtils.hasText(clientId) || !StringUtils.hasText(deviceId)) {
            log.error("【前端取消订阅】clientId或deviceId不能为空");
            throw new IllegalArgumentException("clientId和deviceId不能为空");
        }

        // 从Redis删除订阅关系
        stringRedisTemplate.opsForSet().remove("sub:" + deviceId, clientId);
        stringRedisTemplate.opsForSet().remove("subc:" + clientId, deviceId);
        log.info("【前端取消订阅】前端{}取消订阅设备{}成功", clientId, deviceId);
    }

    /**
     * 全量订阅：前端订阅指定用户名下的所有设备（Controller调用）
     * @param clientId 前端唯一标识（如web_001、app_002）
     * @return 订阅成功的设备数量
     */
    public int subscribeAllDeviceByUserId(String clientId) {
        // 1. 入参校验
        if (!StringUtils.hasText(clientId)) {
            log.error("【全量订阅】clientId不能为空");
            throw new IllegalArgumentException("clientId不能为空");
        }


        // 2. Redis连接可用性校验
        try {
            stringRedisTemplate.hasKey("test:connection");
        } catch (Exception e) {
            log.warn("【全量订阅】Redis连接不可用，订阅操作跳过：{}", e.getMessage());
            return 0;
        }
        Long userId = SecurityUtils.getLoginUser().getUserId();
        boolean update = sysUserService.lambdaUpdate()
                .eq(SysUser::getUserId, userId)
                .set(SysUser::getClientId, clientId)
                .update();
        if (update) {
            log.info("【全量订阅】用户端mqtt客户端绑定成功！");
        }

        // 3. 查询该用户名下的所有设备ID（替换为你的实际设备查询逻辑）
        List<String> deviceIds = new ArrayList<>(agriInfoService.queryImeiByUserId(userId));
        if (userId == 1) {
            deviceIds.add("862538065276061");
        }
        if (deviceIds == null || deviceIds.isEmpty()) {
            log.warn("【全量订阅】用户{}名下无可用设备", userId);
            return 0;
        }
        // 过滤空设备ID，避免无效操作
        List<String> validDeviceIds = deviceIds.stream()
                .filter(StringUtils::hasText)
                .distinct()
                .collect(Collectors.toList());
        if (validDeviceIds.isEmpty()) {
            log.warn("【全量订阅】用户{}名下无有效设备ID", userId);
            return 0;
        }

        // 4. 批量写入Redis订阅关系（兼容JDK 8的RedisCallback写法）
        try {
            stringRedisTemplate.execute(new RedisCallback<Void>() {
                @Override
                public Void doInRedis(RedisConnection connection) throws DataAccessException {
                    // 获取String序列化器（和stringRedisTemplate保持一致）
                    RedisSerializer<String> serializer = stringRedisTemplate.getStringSerializer();

                    // 开启Redis事务
                    connection.multi();
                    byte[] clientIdBytes = serializer.serialize(clientId);
                    // 4.1 设备→前端：给每个设备的订阅集合添加clientId
                    for (String deviceId : validDeviceIds) {
                        byte[] subKey = serializer.serialize("sub:" + deviceId);
                        connection.sAdd(subKey, clientIdBytes);
                    }

                    // 4.2 前端→设备：给前端的订阅集合批量添加所有设备ID
                    byte[] subcKey = serializer.serialize("subc:" + clientId);
                    byte[][] deviceIdBytesArray = new byte[validDeviceIds.size()][];
                    for (int i = 0; i < validDeviceIds.size(); i++) {
                        deviceIdBytesArray[i] = serializer.serialize(validDeviceIds.get(i));
                    }
                    connection.sAdd(subcKey, deviceIdBytesArray);

                    // 新增：给subc设置TTL（兜底“取消订阅失败/异常退出”）
                    connection.expire(subcKey, subcTtlSeconds);

                    // 执行事务
                    connection.exec();
                    return null;
                }
            });

            log.info("【全量订阅】前端{}成功订阅用户{}名下的{}个设备，设备列表：{}",
                    clientId, userId, validDeviceIds.size(), validDeviceIds);
            return validDeviceIds.size();
        } catch (Exception e) {
            log.error("【全量------订阅】前端{}订阅用户{}名下设备失败", clientId, userId, e);
            throw new RuntimeException("全量订阅失败：" + e.getMessage());
        }
    }

    /**
     * 全量取消：前端取消订阅的所有设备（即用户名下所有设备）
     * @param clientId 前端唯一标识
     * @return 需要前端取消监听的MQTT主题列表
     */
    public List<String> unsubscribeAllDevice(String clientId) {
        // 1. 入参校验
        if (!StringUtils.hasText(clientId)) {
            log.error("【全量取消】clientId不能为空");
            throw new IllegalArgumentException("clientId不能为空");
        }

        // 2. Redis连接可用性校验
        try {
            stringRedisTemplate.hasKey("test:connection");
        } catch (Exception e) {
            log.warn("【全量取消】Redis连接不可用，取消操作跳过：{}", e.getMessage());
            return Collections.emptyList();
        }

        // 3. 查询该前端订阅的所有设备ID（即用户名下所有设备）
        Set<String> deviceSet = stringRedisTemplate.opsForSet().members("subc:" + clientId);
        if (deviceSet == null || deviceSet.isEmpty()) {
            log.warn("【全量取消】前端{}无订阅的设备", clientId);
            return Collections.emptyList();
        }

        // 4. 构建需要取消的MQTT主题列表
        List<String> frontendTopics = new ArrayList<>();
        for (String deviceId : deviceSet) {
            frontendTopics.add("frontend/" + clientId + "/dtu/" + deviceId + "/listener");
        }

        // 5. 批量删除Redis订阅关系（兼容JDK 8的RedisCallback写法）
        try {
            stringRedisTemplate.execute(new RedisCallback<Void>() {
                @Override
                public Void doInRedis(RedisConnection connection) throws DataAccessException {
                    RedisSerializer<String> serializer = stringRedisTemplate.getStringSerializer();

                    // 开启事务
                    connection.multi();
                    byte[] clientIdBytes = serializer.serialize(clientId);
                    // 5.1 批量删除设备→前端的订阅关系
                    for (String deviceId : deviceSet) {
                        byte[] subKey = serializer.serialize("sub:" + deviceId);
                        connection.sRem(subKey, clientIdBytes);
                    }

                    // 5.2 删除前端→设备的反向索引（核心：清空该前端的所有订阅设备）
                    byte[] subcKey = serializer.serialize("subc:" + clientId);
                    connection.del(subcKey);

                    // 执行事务
                    connection.exec();
                    return null;
                }
            });
        } catch (Exception e) {
            log.error("【全量取消】Redis批量删除失败", e);
            throw new RuntimeException("全量取消订阅失败：" + e.getMessage());
        }
        log.info("【全量取消】前端{}成功取消{}个设备的订阅", clientId, deviceSet.size());
        return frontendTopics;
    }


}