package cn.iocoder.yudao.module.iot.service.rule.data.action;

import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.ObjUtil;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.RemovalListener;
import lombok.extern.slf4j.Slf4j;

import java.time.Duration;

/**
 * 可缓存的 {@link IotDataRuleAction} 抽象实现
 *
 * 该类提供了一个通用的缓存机制，用于管理各类数据桥接的生产者(Producer)实例。
 *
 * 主要特点:
 * - 基于Guava Cache实现高效的生产者实例缓存管理
 * - 自动处理生产者的生命周期（创建、获取、关闭）
 * - 支持 30 分钟未访问自动过期清理机制
 * - 异常处理与日志记录，便于问题排查
 *
 * 子类需要实现:
 * - initProducer(Config) - 初始化特定类型的生产者实例
 * - closeProducer(Producer) - 关闭生产者实例并释放资源
 *
 * @param <Config>   配置信息类型，用于初始化生产者
 * @param <Producer> 生产者类型，负责将数据发送到目标系统
 * @author HUIHUI
 */
@Slf4j
public abstract class IotDataRuleCacheableAction<Config, Producer> implements IotDataRuleAction {

    /**
     * Producer 缓存
     */
    private final LoadingCache<Config, Producer> PRODUCER_CACHE = CacheBuilder.newBuilder()
            .expireAfterAccess(Duration.ofMinutes(30)) // 30 分钟未访问就提前过期
            .removalListener((RemovalListener<Config, Producer>) notification -> {
                Producer producer = notification.getValue();
                try {
                    closeProducer(producer);
                    log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已关闭]", notification.getKey());
                } catch (Exception e) {
                    log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 关闭失败]", notification.getKey(), e);
                }
            })
            .build(new CacheLoader<Config, Producer>() {

                @Override
                public Producer load(Config config) throws Exception {
                    try {
                        Producer producer = initProducer(config);
                        log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已创建并启动]", config);
                        return producer;
                    } catch (Exception e) {
                        log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 创建启动失败]", config, e);
                        throw e; // 抛出异常，触发缓存加载失败机制
                    }
                }

            });

    /**
     * 获取生产者
     *
     * @param config 配置信息
     * @return 生产者对象
     */
    protected Producer getProducer(Config config) throws Exception {
        return PRODUCER_CACHE.get(config);
    }

    /**
     * 使指定配置的 Producer 缓存失效
     *
     * 当子类检测到 Producer 不可用时（如连接断开），可调用此方法踢出缓存，
     * 下次 {@link #getProducer(Object)} 调用将触发 {@link #initProducer(Object)} 重新创建。
     *
     * @param config 配置信息
     */
    protected void invalidateProducer(Config config) {
        PRODUCER_CACHE.invalidate(config);
    }

    /**
     * 初始化生产者
     *
     * @param config 配置信息
     * @return 生产者对象
     * @throws Exception 如果初始化失败
     */
    protected abstract Producer initProducer(Config config) throws Exception;

    /**
     * 关闭生产者
     *
     * @param producer 生产者对象
     */
    protected abstract void closeProducer(Producer producer) throws Exception;

    @Override
    @SuppressWarnings({"unchecked"})
    public void execute(IotDeviceMessage message, IotDataSinkDO dataSink) {
        Assert.isTrue(ObjUtil.equal(dataSink.getType(), getType()), "类型({})不匹配", dataSink.getType());
        try {
            execute(message, (Config) dataSink.getConfig());
        } catch (Exception e) {
            log.error("[execute][桥梁配置 config({}) 对应的 message({}) 发送异常]", dataSink.getConfig(), message, e);
        }
    }

    /**
     * 执行数据流转
     *
     * @param message 设备消息
     * @param config  配置信息
     */
    protected abstract void execute(IotDeviceMessage message, Config config) throws Exception;

}