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

import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkKafkaConfig;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Kafka 的 {@link IotDataRuleAction} 实现类
 *
 * @author HUIHUI
 */
@ConditionalOnClass(name = "org.springframework.kafka.core.KafkaTemplate")
@Component
@Slf4j
public class IotKafkaDataRuleAction extends
        IotDataRuleCacheableAction<IotDataSinkKafkaConfig, KafkaTemplate<String, String>> {

    private static final Duration SEND_TIMEOUT = Duration.ofSeconds(10);

    @Override
    public Integer getType() {
        return IotDataSinkTypeEnum.KAFKA.getType();
    }

    @Override
    public void execute(IotDeviceMessage message, IotDataSinkKafkaConfig config) throws Exception {
        try {
            // 1. 获取或创建 KafkaTemplate
            KafkaTemplate<String, String> kafkaTemplate = getProducer(config);

            // 2. 发送消息并等待结果
            SendResult<String, String> sendResult = kafkaTemplate.send(config.getTopic(), JsonUtils.toJsonString(message))
                    .get(SEND_TIMEOUT.getSeconds(), TimeUnit.SECONDS);
            // 3. 处理发送结果
            if (sendResult != null && sendResult.getRecordMetadata() != null) {
                log.info("[execute][message({}) config({}) 发送成功，结果: partition={}, offset={}, timestamp={}]",
                        message, config,
                        sendResult.getRecordMetadata().partition(),
                        sendResult.getRecordMetadata().offset(),
                        sendResult.getRecordMetadata().timestamp());
            } else {
                log.warn("[execute][message({}) config({}) 发送结果为空]", message, config);
            }
        } catch (Exception e) {
            log.error("[execute][message({}) config({}) 发送失败]", message, config, e);
            throw e;
        }
    }

    @Override
    protected KafkaTemplate<String, String> initProducer(IotDataSinkKafkaConfig config) {
        // 1.1 构建生产者配置
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 1.2 如果配置了认证信息
        if (config.getUsername() != null && config.getPassword() != null) {
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.mechanism", "PLAIN");
            props.put("sasl.jaas.config",
                    "org.apache.kafka.common.security.plain.PlainLoginModule required username=\""
                            + config.getUsername() + "\" password=\"" + config.getPassword() + "\";");
        }
        // 1.3 如果启用 SSL
        if (Boolean.TRUE.equals(config.getSsl())) {
            props.put("security.protocol", "SSL");
        }

        // 2. 创建 KafkaTemplate
        DefaultKafkaProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(props);
        return new KafkaTemplate<>(producerFactory);
    }

    @Override
    protected void closeProducer(KafkaTemplate<String, String> producer) {
        producer.destroy();
    }

}