package com.inspect.analysis.service.impl;

import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.google.common.util.concurrent.RateLimiter;
import com.inspect.analysis.service.RetryableRequest;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.data.redis.core.RedisTemplate;

import java.time.Duration;
import java.util.concurrent.*;
import java.util.function.Consumer;

@Slf4j
public class RedisQueueRetryableConsumerAsync<T> implements Runnable {

    private final RedisTemplate<String, String> redisTemplate;
    private final Gson gson = new Gson();
    private final String queueKey;
    private final Class<T> clazz;

    private final long idleSleepMs;

    private final Consumer<T> handler;

    private volatile boolean running = false;

    private final RateLimiter rateLimiter;

    public ThreadPoolExecutor getExecutor() {
        return executor;
    }

    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(
            4, // core threads
            10, // max threads
            60,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(100), // 队列容量
            new ThreadPoolExecutor.CallerRunsPolicy() // 背压策略：调用者线程执行任务
    );

    public RedisQueueRetryableConsumerAsync(RedisTemplate<String, String> redisTemplate,
                                            String queueKey,
                                            Class<T> clazz,
                                            double permitsPerSecond, // 每秒许可数，控制速率
                                            long idleSleepMs,
                                            Consumer<T> handler) {
        this.redisTemplate = redisTemplate;
        this.queueKey = queueKey;
        this.clazz = clazz;
        this.idleSleepMs = idleSleepMs;
        this.handler = handler;
        this.rateLimiter = RateLimiter.create(permitsPerSecond);
    }

    public void start() {
        if (!running) {
            running = true;
            new Thread(this, "RedisQueueRetryableConsumerAsync-" + queueKey).start();
        }
    }

    public void stop() {
        running = false;
        executor.shutdown();
    }

    @Override
    public void run() {
        while (running) {
            try {
                String json = redisTemplate.opsForList().leftPop(queueKey, Duration.ofSeconds(1));
                if (StringUtils.isEmpty(json)) {
                    Thread.sleep(idleSleepMs);
                    continue;
                }

                rateLimiter.acquire();

                T obj = gson.fromJson(json, clazz);

                // 判断线程池是否拥堵
                if (executor.getQueue().size() > 80) {
                    log.warn("RedisQueueRetryableConsumerAsync Thread pool queue is almost full, pushing back to Redis: {}", queueKey);
                    // 推回 Redis 保证不丢数据
                    redisTemplate.opsForList().rightPush(queueKey, json);
                    Thread.sleep(1000);
                    continue;
                }

                executor.submit(() -> {
                    try {
                        handler.accept(obj);
                    } catch (Exception e) {
                        log.warn("RedisQueueRetryableConsumerAsync handler failed for obj={}, retrying...", json, e);

                        // 反序列化带 retryCount 的对象
                        if (obj instanceof RetryableRequest) {
                            RetryableRequest rr = (RetryableRequest) obj;
                            int retry = rr.getRetryCount();
                            if (retry < 3) {
                                rr.setRetryCount(retry + 1);
                                redisTemplate.opsForList().rightPush(queueKey, gson.toJson(rr));
                            } else {
                                log.error("RedisQueueRetryableConsumerAsync max retries reached for: {}", json);
                            }
                        } else {
                            redisTemplate.opsForList().rightPush(queueKey, json); // 简单重入队
                        }
                    }
                });
            } catch (Exception e) {
                log.error("RedisQueueRetryableConsumerAsync error", e);
            }
        }

        log.info("RedisQueueRetryableConsumerAsync for [{}] stopped.", queueKey);
    }
}

