package com.fs.live.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.fs.common.core.domain.R;
import com.fs.common.core.redis.RedisCache;
import com.fs.common.utils.DateUtils;
import com.fs.common.utils.spring.SpringUtils;
import com.fs.live.domain.LiveUserBehaviorTrack;
import com.fs.live.mapper.LiveUserBehaviorTrackMapper;
import com.fs.live.service.ILiveUserBehaviorTrackService;
import com.fs.live.util.LiveBehaviorTrackUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;

/**
 * 直播用户行为轨迹Service业务层处理
 *
 * @author xw
 * @date 2026-01-05
 */
@Slf4j
@Service
public class LiveUserBehaviorTrackServiceImpl implements ILiveUserBehaviorTrackService {

    @Autowired
    private LiveUserBehaviorTrackMapper liveUserBehaviorTrackMapper;

    @Autowired(required = false)
    private RedisCache redisCache;

    private static final String BEHAVIOR_BUFFER_KEY = "live:behavior:buffer:";

    private static final int PUSH_THRESHOLD = 50;

    private static final long WINDOW_MS = 100;

    // 直播间最后推送时间缓存
    private static final ConcurrentHashMap<Long, Long> lastPushTime = new ConcurrentHashMap<>();
    // 推送任务调度器
    private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    // 推送执行器（异步非阻塞）
    private static final ExecutorService pushExecutor = Executors.newFixedThreadPool(5);

    /**
     * 查询直播用户行为轨迹
     *
     * @param id 主键
     * @param userId 用户ID（用于计算表后缀）
     * @return 直播用户行为轨迹
     */
    @Override
    public LiveUserBehaviorTrack selectLiveUserBehaviorTrackById(Long id, Long userId) {
        String tableSuffix = LiveBehaviorTrackUtil.getTableSuffix(userId);
        return liveUserBehaviorTrackMapper.selectLiveUserBehaviorTrackById(id, tableSuffix);
    }

    /**
     * 查询直播用户行为轨迹列表
     *
     * @param track 直播用户行为轨迹
     * @return 直播用户行为轨迹集合
     */
    @Override
    public List<LiveUserBehaviorTrack> selectLiveUserBehaviorTrackList(LiveUserBehaviorTrack track) {
        // 如果有userId，直接查询对应的分表
        if (track.getUserId() != null) {
            String tableSuffix = LiveBehaviorTrackUtil.getTableSuffix(track.getUserId());
            track.setTableSuffix(tableSuffix);
            return liveUserBehaviorTrackMapper.selectLiveUserBehaviorTrackList(track);
        }

        // 如果没有userId，需要查询所有分表并聚合结果
        List<LiveUserBehaviorTrack> resultList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            track.setTableSuffix(String.valueOf(i));
            List<LiveUserBehaviorTrack> list = liveUserBehaviorTrackMapper.selectLiveUserBehaviorTrackList(track);
            if (list != null && !list.isEmpty()) {
                resultList.addAll(list);
            }
        }

        // 按时间倒序排序
        resultList.sort((a, b) -> {
            if (a.getBehaviorTime() == null || b.getBehaviorTime() == null) {
                return 0;
            }
            return b.getBehaviorTime().compareTo(a.getBehaviorTime());
        });

        return resultList;
    }

    /**
     * 新增直播用户行为轨迹（同步）
     *
     * @param track 直播用户行为轨迹
     * @return 结果
     */
    @Override
    public int insertLiveUserBehaviorTrack(LiveUserBehaviorTrack track) {
        // 自动设置分表后缀
        if (track.getTableSuffix() == null) {
            track.setTableSuffix(LiveBehaviorTrackUtil.getTableSuffix(track.getUserId()));
        }
        // 设置创建时间
        if (track.getCreateTime() == null) {
            track.setCreateTime(DateUtils.getNowDate());
        }
        // 设置行为时间戳
        if (track.getBehaviorTime() == null) {
            track.setBehaviorTime(System.currentTimeMillis());
        }

        return liveUserBehaviorTrackMapper.insertLiveUserBehaviorTrack(track);
    }

    /**
     *异步+实时
     * @param track 直播用户行为轨迹
     */
    @Async
    @Override
    public void asyncInsertLiveUserBehaviorTrack(LiveUserBehaviorTrack track) {
        try {
            insertLiveUserBehaviorTrack(track);

            pushToScreenRealtime(track);

        } catch (Exception e) {
            log.error("异步插入用户行为轨迹失败: userId={}, liveId={}, behaviorType={}",
                     track.getUserId(), track.getLiveId(), track.getBehaviorType(), e);
        }
    }

    /**
     * 实时推送到大屏（滑动窗口批量合并方案）
     *
     * 策略：
     * 1. 行为先缓存到 Redis（极快）
     * 2. 达到阈值（50条）立即推送
     * 3. 未达到阈值，100ms 后自动推送
     * 4. 自动削峰填谷，处理高并发
     *
     * @param track 用户行为轨迹
     */
    private void pushToScreenRealtime(LiveUserBehaviorTrack track) {
        if (redisCache == null) {
            return;
        }

        try {
            Long liveId = track.getLiveId();
            String bufferKey = BEHAVIOR_BUFFER_KEY + liveId;

            // 构建精简数据
            java.util.Map<String, Object> screenData = new java.util.HashMap<>();
            screenData.put("userId", track.getUserId());
            screenData.put("behaviorType", track.getBehaviorType());
            screenData.put("behaviorDesc", track.getBehaviorDesc());
            screenData.put("behaviorTime", track.getBehaviorTime());
            if (track.getResourceId() != null) {
                screenData.put("resourceId", track.getResourceId());
            }

            redisCache.leftPushList(bufferKey, JSONObject.toJSONString(screenData));
            long bufferSize = redisCache.getListSize(bufferKey);

            // 1：达到阈值，立即推送
            if (bufferSize >= PUSH_THRESHOLD) {
                triggerPush(liveId, false);
            }
            // 2：未达到阈值，调度延迟推送
            else {
                scheduleDelayedPush(liveId);
            }

        } catch (Exception e) {
            log.debug("[实时推送] 推送失败, liveId={}, userId={}",
                    track.getLiveId(), track.getUserId());
        }
    }

    /**
     * 调度延迟推送（滑动窗口）
     * 100ms 内如果没有新的推送触发，则自动推送当前缓冲区
     */
    private void scheduleDelayedPush(Long liveId) {
        Long now = System.currentTimeMillis();
        Long last = lastPushTime.get(liveId);

        // 如果最近刚推送过，不重复调度
        if (last != null && (now - last) < WINDOW_MS) {
            return;
        }

        // 调度延迟推送任务
        scheduler.schedule(() -> {
            triggerPush(liveId, true);
        }, WINDOW_MS, java.util.concurrent.TimeUnit.MILLISECONDS);
    }

    /**
     * 触发推送（从缓冲区读取并推送到 WebSocket）
     *
     * @param liveId 直播间ID
     * @param isDelayed 是否是延迟触发（用于日志）
     */
    private void triggerPush(Long liveId, boolean isDelayed) {
        // 异步执行推送，不阻塞
        pushExecutor.submit(() -> {
            try {
                String bufferKey = BEHAVIOR_BUFFER_KEY + liveId;

                // 从缓冲区读取所有数据（一次性清空）
                List<String> behaviorList = new ArrayList<>();
                while (true) {
                    String behavior = redisCache.rightPopList(bufferKey);
                    if (behavior == null) {
                        break;
                    }
                    behaviorList.add(behavior);
                }

                if (behaviorList.isEmpty()) {
                    return;
                }

                pushToWebSocket(liveId, behaviorList);

                // 更新最后推送时间
                lastPushTime.put(liveId, System.currentTimeMillis());

                log.info("[实时推送] 推送成功, liveId={}, count={}, delayed={}",
                        liveId, behaviorList.size(), isDelayed);

            } catch (Exception e) {
                log.error("[实时推送] 推送失败, liveId={}", liveId, e);
            }
        });
    }

    /**
     * 推送到 WebSocket（支持本地和HTTP转发）
     */
    private void pushToWebSocket(Long liveId, List<String> behaviorList) {
        try {
            boolean hasBean = SpringUtils.containsBean("webSocketServer");

            if (hasBean) {
                pushToLocalWebSocket(liveId, behaviorList);
            } else {
                pushToRemoteWebSocket(liveId, behaviorList);
            }

        } catch (Exception e) {
            log.error("[实时推送] WebSocket推送失败, liveId={}", liveId, e);
        }
    }

    /**
     * 本地推送：直接调用WebSocketServer
     */
    private void pushToLocalWebSocket(Long liveId, List<String> behaviorList) {
        try {
            Object webSocketServer = SpringUtils.getBean("webSocketServer");
            if (webSocketServer == null) {
                return;
            }

            // 构建推送消息
            Object sendMsgVo = buildBatchMessage(liveId, behaviorList);
            String message = JSONObject.toJSONString(R.ok().put("data", sendMsgVo));

            // 调用 broadcastMessage 推送
            Method broadcastMethod = webSocketServer.getClass()
                .getMethod("broadcastMessage", Long.class, String.class);
            broadcastMethod.invoke(webSocketServer, liveId, message);

            log.info("[实时推送] 本地WebSocket推送成功, liveId={}, count={}", liveId, behaviorList.size());

        } catch (Exception e) {
            log.error("[实时推送] 本地WebSocket推送失败, liveId={}", liveId, e);
        }
    }

    /**
     * 远程推送：通过HTTP转发到fs-live-app
     */
    private void pushToRemoteWebSocket(Long liveId, List<String> behaviorList) {
        try {
            // 读取配置的fs-live-app地址
            Environment env = SpringUtils.getBean(Environment.class);
            String liveWebSocketUrl = env.getProperty("liveWebSocketUrl");

            if (liveWebSocketUrl == null || liveWebSocketUrl.isEmpty()) {
                log.warn("[实时推送] 未配置liveWebSocketUrl，跳过远程推送");
                return;
            }

            log.info("[实时推送] 使用远程模式推送, url={}", liveWebSocketUrl);

            Map<String, Object> params = new HashMap<>();
            params.put("liveId", liveId);
            params.put("behaviors", behaviorList);

            String url = liveWebSocketUrl + "/app/live/behavior/push";
            String jsonParams = JSONObject.toJSONString(params);

            // 使用RestTemplate发送POST请求
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);

            HttpEntity<String> request = new HttpEntity<>(jsonParams, headers);

            RestTemplate restTemplate = new RestTemplate();

            ResponseEntity<String> response = restTemplate.postForEntity(url, request, String.class);

            log.info("[实时推送] 远程HTTP转发成功, liveId={}, count={}, status={}",
                    liveId, behaviorList.size(), response.getStatusCode());

        } catch (Exception e) {
            log.error("[实时推送] 远程HTTP转发失败, liveId={}", liveId, e);
        }
    }

    /**
     * 构建批量推送消息
     */
    private Object buildBatchMessage(Long liveId, List<String> behaviorList) {
        try {
            Class<?> sendMsgVoClass = Class.forName("com.fs.live.websocket.bean.SendMsgVo");
            Object sendMsgVo = sendMsgVoClass.newInstance();

            // 解析为对象列表
            List<Map<String, Object>> behaviors = new ArrayList<>();
            for (String behaviorStr : behaviorList) {
                try {
                    Map<String, Object> behavior = JSONObject.parseObject(behaviorStr, Map.class);
                    behaviors.add(behavior);
                } catch (Exception e) {
                    log.warn("[实时推送] 解析行为数据失败: {}", behaviorStr);
                }
            }

            sendMsgVoClass.getMethod("setLiveId", Long.class).invoke(sendMsgVo, liveId);
            sendMsgVoClass.getMethod("setCmd", String.class).invoke(sendMsgVo, "behaviorTrackBatch");
            sendMsgVoClass.getMethod("setMsg", String.class).invoke(sendMsgVo, "用户行为实时更新");
            sendMsgVoClass.getMethod("setData", String.class).invoke(sendMsgVo, JSONObject.toJSONString(behaviors));

            return sendMsgVo;
        } catch (Exception e) {
            log.warn("[实时推送] 构建消息失败", e);
            return buildFallbackBatchMessage(liveId, behaviorList);
        }
    }

    /**
     * 构建备用批量消息
     */
    private Map<String, Object> buildFallbackBatchMessage(Long liveId, List<String> behaviorList) {
        Map<String, Object> msg = new HashMap<>();
        msg.put("liveId", liveId);
        msg.put("cmd", "behaviorTrackBatch");
        msg.put("msg", "用户行为实时更新");
        msg.put("data", "[" + String.join(",", behaviorList) + "]");
        return msg;
    }

    /**
     * 批量新增直播用户行为轨迹
     *
     * @param trackList 直播用户行为轨迹列表
     * @return 结果
     */
    @Override
    public int batchInsertLiveUserBehaviorTrack(List<LiveUserBehaviorTrack> trackList) {
        if (trackList == null || trackList.isEmpty()) {
            return 0;
        }

        // 按分表后缀分组
        Map<String, List<LiveUserBehaviorTrack>> groupedMap = trackList.stream()
                .peek(track -> {
                    // 自动设置分表后缀
                    if (track.getTableSuffix() == null) {
                        track.setTableSuffix(LiveBehaviorTrackUtil.getTableSuffix(track.getUserId()));
                    }
                    // 设置创建时间
                    if (track.getCreateTime() == null) {
                        track.setCreateTime(DateUtils.getNowDate());
                    }
                    // 设置行为时间戳
                    if (track.getBehaviorTime() == null) {
                        track.setBehaviorTime(System.currentTimeMillis());
                    }
                })
                .collect(Collectors.groupingBy(LiveUserBehaviorTrack::getTableSuffix));

        int totalCount = 0;
        // 分表批量插入
        for (Map.Entry<String, List<LiveUserBehaviorTrack>> entry : groupedMap.entrySet()) {
            try {
                int count = liveUserBehaviorTrackMapper.batchInsertLiveUserBehaviorTrack(entry.getValue());
                totalCount += count;
            } catch (Exception e) {
                log.error("批量插入用户行为轨迹失败: tableSuffix={}", entry.getKey(), e);
            }
        }

        return totalCount;
    }

    /**
     * 修改直播用户行为轨迹
     *
     * @param track 直播用户行为轨迹
     * @return 结果
     */
    @Override
    public int updateLiveUserBehaviorTrack(LiveUserBehaviorTrack track) {
        // 自动设置分表后缀
        if (track.getTableSuffix() == null) {
            track.setTableSuffix(LiveBehaviorTrackUtil.getTableSuffix(track.getUserId()));
        }
        return liveUserBehaviorTrackMapper.updateLiveUserBehaviorTrack(track);
    }

    /**
     * 批量删除直播用户行为轨迹
     *
     * @param ids 需要删除的数据主键集合
     * @param userId 用户ID（用于计算表后缀）
     * @return 结果
     */
    @Override
    public int deleteLiveUserBehaviorTrackByIds(Long[] ids, Long userId) {
        String tableSuffix = LiveBehaviorTrackUtil.getTableSuffix(userId);
        return liveUserBehaviorTrackMapper.deleteLiveUserBehaviorTrackByIds(ids, tableSuffix);
    }

    /**
     * 删除直播用户行为轨迹信息
     *
     * @param id 主键
     * @param userId 用户ID（用于计算表后缀）
     * @return 结果
     */
    @Override
    public int deleteLiveUserBehaviorTrackById(Long id, Long userId) {
        String tableSuffix = LiveBehaviorTrackUtil.getTableSuffix(userId);
        return liveUserBehaviorTrackMapper.deleteLiveUserBehaviorTrackById(id, tableSuffix);
    }

    /**
     * 根据用户ID和直播ID查询行为轨迹
     *
     * @param userId 用户ID
     * @param liveId 直播ID
     * @return 行为轨迹列表
     */
    @Override
    public List<LiveUserBehaviorTrack> selectByUserIdAndLiveId(Long userId, Long liveId) {
        String tableSuffix = LiveBehaviorTrackUtil.getTableSuffix(userId);
        return liveUserBehaviorTrackMapper.selectByUserIdAndLiveId(userId, liveId, tableSuffix);
    }

    /**
     * 根据直播ID统计各类行为数量（聚合所有分表数据）
     *
     * @param liveId 直播ID
     * @return 统计结果
     */
    @Override
    public Map<String, Long> statisticsByLiveId(Long liveId) {
        Map<String, Long> resultMap = new HashMap<>();

        // 查询所有分表并聚合结果
        for (int i = 0; i < 10; i++) {
            String tableSuffix = String.valueOf(i);
            try {
                List<Map<String, Object>> list = liveUserBehaviorTrackMapper.statisticsByLiveId(liveId, tableSuffix);
                if (list != null && !list.isEmpty()) {
                    for (Map<String, Object> map : list) {
                        String behaviorDesc = (String) map.get("behaviorDesc");
                        Long count = ((Number) map.get("count")).longValue();
                        resultMap.merge(behaviorDesc, count, Long::sum);
                    }
                }
            } catch (Exception e) {
                log.error("统计直播行为失败: liveId={}, tableSuffix={}", liveId, tableSuffix, e);
            }
        }

        return resultMap;
    }

    /**
     * 根据用户ID统计行为数量
     *
     * @param userId 用户ID
     * @return 统计结果
     */
    @Override
    public Map<String, Long> statisticsByUserId(Long userId) {
        Map<String, Long> resultMap = new HashMap<>();
        String tableSuffix = LiveBehaviorTrackUtil.getTableSuffix(userId);

        try {
            List<Map<String, Object>> list = liveUserBehaviorTrackMapper.statisticsByUserId(userId, tableSuffix);
            if (list != null && !list.isEmpty()) {
                for (Map<String, Object> map : list) {
                    String behaviorDesc = (String) map.get("behaviorDesc");
                    Long count = ((Number) map.get("count")).longValue();
                    resultMap.put(behaviorDesc, count);
                }
            }
        } catch (Exception e) {
            log.error("统计用户行为失败: userId={}", userId, e);
        }

        return resultMap;
    }

    /**
     * 根据直播ID统计各类行为数量（指定分表）
     *
     * @param liveId 直播ID
     * @param tableSuffix 表后缀
     * @return 统计结果
     */
    @Override
    public List<Map<String, Object>> statisticsByLiveIdWithTable(Long liveId, String tableSuffix) {
        return liveUserBehaviorTrackMapper.statisticsByLiveId(liveId, tableSuffix);
    }
}
