package com.ard.work.component.websocket.handler;

import com.alibaba.fastjson2.JSON;
import com.ard.work.domain.message.BaseMessage;
import com.ard.work.component.websocket.utils.WebSocketUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

@Slf4j
public abstract class BaseSubscriptionManager<T> {

    // 存储订阅的用户
    protected final Map<String, Set<Long>> subscriptions = new ConcurrentHashMap<>();
    // 定时任务调度器
    protected final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10);
    // 存储推送任务
    protected final Map<String, ScheduledFuture<?>> pushTasks = new ConcurrentHashMap<>();

    /**
     * 订阅指定设备的数据
     */
    public void subscribe(String boxSn, Long userId, Integer period) {
        subscriptions.computeIfAbsent(boxSn, k -> new HashSet<>()).add(userId);
        if (subscriptions.get(boxSn).size() == 1) {
            startPushTask(boxSn, period);
        }
    }

    /**
     * 取消订阅
     */
    public void unsubscribe(String boxSn, Long userId) {
        Set<Long> userIds = subscriptions.get(boxSn);
        if (userIds != null) {
            userIds.remove(userId);
            if (userIds.isEmpty()) {
                stopPushTask(boxSn);
            }
        }
    }

    /**
     * 启动推送任务
     */
    private void startPushTask(String boxSn, Integer period) {
        // 如果已经存在任务，则不再创建
        if (pushTasks.containsKey(boxSn)) {
            return;
        }
        ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
            sendDataToSubscribers(boxSn);
        }, 0, period, TimeUnit.SECONDS);

        pushTasks.put(boxSn, future);
    }

    /**
     * 停止推送任务
     */
    private void stopPushTask(String boxSn) {
        ScheduledFuture<?> future = pushTasks.get(boxSn);
        if (future != null) {
            future.cancel(false);
        }
        pushTasks.remove(boxSn);
    }

    /**
     * 移除用户订阅
     */
    public void removeSubscribe(Long userId) {
        for (Set<Long> userIds : subscriptions.values()) {
            userIds.remove(userId);
        }
    }


    /**
     * 向订阅者发送数据
     */
    private void sendDataToSubscribers(String boxSn) {
        T data = getData(boxSn);
        BaseMessage<T> baseMessage = new BaseMessage<>(getMessageType(), data);
        Set<Long> userIds = subscriptions.get(boxSn);
        if (userIds != null) {
            for (Long userId : userIds) {
                try {
                    log.debug("给用户{}定时推送", userId);
                    WebSocketUtil.sendMessage(userId, JSON.toJSONString(baseMessage));
                } catch (Exception e) {
                    log.error("Error sending message to session {}", userId, e);
                }
            }
        }
    }

    /**
     * 获取数据的具体实现
     */
    protected abstract T getData(String boxSn);

    /**
     * 获取消息类型
     */
    protected abstract String getMessageType();
}