package com.huashe.park.application.engine;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

import javax.annotation.Resource;


import com.huashe.park.core.redis.RedisProxy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import com.alibaba.fastjson2.JSONObject;
import com.huashe.common.utils.StringUtils;
import com.huashe.park.common.DateTimeUtil;

public abstract class EvtFusionEngine implements IFusionEngine {
    private final static Logger logger = LoggerFactory.getLogger(EvtFusionEngine.class);

    @Resource
    private RedisProxy redisService;

    private String engineName;

    @Value("${evt-fusion.thread-pool-size:2}")
    private int threadPoolSize;

    @Resource(name = "threadPoolTaskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    public void init() {
        List<LinkedBlockingQueue<JSONObject>> queue = getQueue();
        for (int i = 0; i < threadPoolSize; i++) {
            queue.add(new LinkedBlockingQueue<>());
        }
    }

    public void setEngineName(String name) {
        this.engineName = name;
    }

    public void push(JSONObject msg) {
        String key = msg.getString("key");
        if (StringUtils.isEmpty(key)) {
            return;
        }
        int bucketIndex = computeHashModulo(key, threadPoolSize);
        List<LinkedBlockingQueue<JSONObject>> queue = getQueue();
        LinkedBlockingQueue<JSONObject> messageQueue = queue.get(bucketIndex);
        messageQueue.offer(msg);
    }

    public void start() {
        List<LinkedBlockingQueue<JSONObject>> queue = getQueue();
        for (int i = 0; i < threadPoolSize; i++) {
            int finalI = i;
            threadPoolTaskExecutor.execute(() -> {
                while (true) {
                    try {
                        JSONObject msg = queue.get(finalI).take();
                        LocationInfo locationInfoNew = new LocationInfo(msg.getDouble("latitude"),
                            msg.getDouble("longitude"), msg.getLong("srcTimestamp"));
                        locationInfoNew.setMsg(msg);
                        String key = getKey(locationInfoNew);
                        if (!redisService.hasKey(key)) {
                            getBizId(locationInfoNew);
                            redisService.setCacheObject(key, locationInfoNew);
                            newEvtCallback(locationInfoNew);
                            continue;
                        }
                        LocationInfo locationInfo = redisService.getCacheObject(key);
                        locationInfoNew.setEvtTimestamp(DateTimeUtil.timestampMillis());
                        // 判断消息是否需要融合
                        if (check(locationInfo, locationInfoNew)) {
                            locationInfo.setSrcTimestamp(locationInfoNew.getSrcTimestamp());
                            locationInfo.setHandleTimestamp(DateTimeUtil.timestampMillis());
                            redisService.setCacheObject(key, locationInfo);
                        }
                        else {
                            // 老数据释放
                            processOlderData(locationInfo);
                            getBizId(locationInfoNew);
                            redisService.setCacheObject(key, locationInfoNew);
                            newEvtCallback(locationInfoNew);
                        }
                    }
                    catch (InterruptedException e) {
                        logger.error("{} error", this.engineName, e);
                        // 重置中断状态
                        Thread.currentThread().interrupt();
                        // 根据业务逻辑决定是否继续执行
                        if (Thread.currentThread().isInterrupted()) {
                            logger.error("任务已中断，不再继续执行");
                            break;
                        }
                    }
                }
            });
        }
    }

    /**
     * 计算字符串的哈希值并对某个数取余数。
     *
     * @param str 字符串
     * @param modulo 取余数的基数
     * @return 字符串哈希值对 modulo 取余的结果
     */
    private int computeHashModulo(String str, int modulo) {
        long hash = 0;
        long base = 31;
        long baseMod = base % modulo; // 预计算乘法因子

        for (char c : str.toCharArray()) {
            hash = (hash * baseMod + c) % modulo; // 避免整数溢出
        }

        return (int) (hash % modulo); // 最终结果转换为 int
    }

}
