package com.inspect.partrolresult.service;

import com.inspect.analysis.constant.AnalyseConstants;
import com.inspect.base.redis.service.RedisService;
import com.inspect.partrolresult.domain.AnalyseRequest;
import lombok.extern.slf4j.Slf4j;
import org.redisson.RedissonShutdownException;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.redisson.client.RedisException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Slf4j
@Component
public class AnalyseRequestDelayQueueListener implements InitializingBean, DisposableBean {
    private final String THREAD_NAME = "AnalyseRequest-DelayQueue-Listener";

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private RedisService redisService;

    @Resource
    private IAnalyseRequestService analyseRequestService;

    private Thread listenerThread;
    private volatile boolean running = true;

    @Override
    public void afterPropertiesSet() {
        // 启动监听线程
        listenerThread = new Thread(() -> {
            RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(AnalyseConstants.ALGORITHM_REQUEST_DELAY_QUEUE);
            log.info("Listening AnalyseRequest Delay Queue Task Launching");

            while (running && !Thread.currentThread().isInterrupted()) {
                try {
//                    AnalyseRequest request = blockingQueue.take(); // 阻塞等待任务到期
//                    handleRequest(request);
                    final String requestId = blockingQueue.take(); // 阻塞等待任务到期
                    log.info("ANALYSE_REQUEST_DELAY_QUEUE_EXPIRE requestId: {}", requestId);
                } catch (RedissonShutdownException e) {
                    log.warn("Redisson 已关闭，终止监听线程");
                    break;
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt(); // 响应中断
                    log.info("监听线程被中断，正在关闭...");
                    break;
                } catch (RedisException e) {
                    log.error("Redis 连接异常: {}", e.getMessage(), e);
                    try {
                        Thread.sleep(3000); // 避免异常频繁重试
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
                catch (Exception e) {
                    log.error("Listening AnalyseRequest Queue Exception: ", e);
                }
            }
        }, THREAD_NAME);

        listenerThread.setDaemon(true); // 设置为守护线程，防止主程序不退出
        listenerThread.start();
    }

    private void handleRequest(AnalyseRequest request) {
        String redisKey = request.isFilter() ? AnalyseConstants.ANALYSE_FILTER_REQUEST.concat(request.getRequestId())
                : AnalyseConstants.ANALYSE_AI_REQUEST.concat(request.getRequestId());
        //Object cached = redisService.redisTemplate.opsForValue().getAndDelete(redisKey);
        Object cached = redisService.redisTemplate.opsForValue().get(redisKey);
        if (cached != null) {
            // 正常到期但业务未处理，走补偿逻辑
            log.info("ANALYSE_REQUEST_DELAY_QUEUE COMPENSATE patrolPointId: {}, requestId: {}, BODY: {}",
                    request.getTaskPatrolId(),
                    request.getRequestId(),
                    request);
            analyseRequestService.sendCompensateRequest(request);
        } else {
            // 正常业务已完成，什么也不做，正常情况下不会进入这个逻辑
            log.info("ANALYSE_REQUEST_DELAY_QUEUE CONSUMED: patrolPointId: {}, requestId: {}",
                    request.getTaskPatrolId(),
                    request.getRequestId());
        }
    }

    @Override
    public void destroy() {
        log.info("正在关闭 AnalyseRequestDelayQueueListener...");
        running = false;
        if (listenerThread != null) {
            listenerThread.interrupt();
        }
    }
}

