package com.infoloop.tianting.logic.delay;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

@Slf4j
@Service
@RequiredArgsConstructor
public class DelayedQueue<T> {

    private final RedissonClient redissonClient;

    private final Map<String, RDelayedQueue<T>> delayedQueues = new ConcurrentHashMap<>();

    /**
     * 初始化队列，传入队列名称和处理逻辑
     */
    public void initQueue(String queueName, Consumer<T> processMessage) {
        RBlockingQueue<T> blockingQueue = redissonClient.getBlockingQueue(queueName);
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
        delayedQueues.put(queueName, delayedQueue);
        DelayTaskQueueExecutor.Processor<T> processor = task -> {
            try {
                processMessage.accept(task);
            } catch (Exception e) {
                log.error("Error processing task: {}", task, e);
            }
        };
        new DelayTaskQueueExecutor<>(queueName + " DELAY TASK", redissonClient, blockingQueue, processor);
        log.info("Initialized delay queue for {}", queueName);
    }

    public void addToQueue(String queueName, T message, long delay, TimeUnit timeUnit) {
        RDelayedQueue<T> queue = getDelayedQueue(queueName);
        queue.offer(message, delay, timeUnit);
        log.info("Added to queue: {}, msg: {}, delay: {} {}", queueName, message, delay, timeUnit);
    }

    public void addToQueue(String queueName, List<T> messages, long delay, TimeUnit timeUnit) {
        RDelayedQueue<T> queue = getDelayedQueue(queueName);
        messages.forEach(message -> {
            queue.offer(message, delay, timeUnit);
            log.info("Added to queue: {}, msg: {}, delay: {} {}", queueName, message, delay, timeUnit);
        });
    }

    private RDelayedQueue<T> getDelayedQueue(String queueName) {
        RDelayedQueue<T> queue = delayedQueues.get(queueName);
        if (queue == null) {
            throw new IllegalArgumentException("Queue not initialized: " + queueName);
        }
        return queue;
    }

}