package study.u48;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.TopicExistsException;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * U48 Spring Kafka：启用配置、监听生命周期、ack 语义、错误处理与事务。
 * 连接真实 Kafka 4.1.2 KRaft 单机 broker（由 runner 以独立进程启动与回收）。
 */
public class KafkaLab {

    static final List<String> CONSUMED = new CopyOnWriteArrayList<>();
    static final List<String> TX_CONSUMED = new CopyOnWriteArrayList<>();
    static final AtomicInteger ERROR_ATTEMPTS = new AtomicInteger();
    static CountDownLatch MAIN_LATCH = new CountDownLatch(1);
    static CountDownLatch MANUAL_LATCH = new CountDownLatch(1);
    static CountDownLatch ERROR_LATCH = new CountDownLatch(1);
    static CountDownLatch TX_LATCH = new CountDownLatch(2);

    public static void main(String[] args) throws Exception {
        try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) {
            context.register(KafkaConfig.class);
            context.refresh();

            KafkaTemplate<String, String> template = context.getBean("kafkaTemplate", KafkaTemplate.class);

            // ---- 1. 启用配置：KafkaAdmin 创建主题，模板发送，监听消费 ----
            template.send("u48-t1", "msg-1").get(10, TimeUnit.SECONDS);
            check(MAIN_LATCH.await(15, TimeUnit.SECONDS), "监听器应消费 msg-1");
            out("kafkaListener topicCreated=true consumed=1");

            // ---- 2. 监听生命周期：stop 后不消费、start 后恢复 ----
            var container = context.getBean(KafkaListenerEndpointRegistry.class)
                    .getListenerContainer("mainListener");
            container.stop();
            check(!container.isRunning(), "容器 stop 后应停止");
            template.send("u48-t1", "msg-2").get(10, TimeUnit.SECONDS);
            int before = CONSUMED.size();
            Thread.sleep(800);
            check(CONSUMED.size() == before, "停止期间不应消费");
            container.start();
            // resume 后等消费
            long deadline = System.currentTimeMillis() + 15_000;
            while (CONSUMED.size() == before && System.currentTimeMillis() < deadline) {
                Thread.sleep(100);
            }
            check(CONSUMED.size() > before, "resume 后应恢复消费");
            out("kafkaLifecycle stoppedNoConsume=true resumed=true");

            // ---- 3. MANUAL ack：不 ack 的记录在容器重启后被重新投递（at-least-once） ----
            // 注意 MANUAL 语义：ack 后一条会连同提交之前的 offset，因此本段只投一条并分阶段验证
            var manualContainer = context.getBean(KafkaListenerEndpointRegistry.class)
                    .getListenerContainer("manualListener");
            MANUAL_LATCH = new CountDownLatch(1);
            template.send("u48-t2", "no-ack").get(10, TimeUnit.SECONDS);
            check(MANUAL_LATCH.await(15, TimeUnit.SECONDS), "manual 监听器应收到 no-ack");
            manualContainer.stop();
            CONSUMED.removeIf(m -> m.startsWith("manual:"));
            MANUAL_LATCH = new CountDownLatch(1);
            manualContainer.start();
            check(MANUAL_LATCH.await(20, TimeUnit.SECONDS), "未 ack 的记录应在重启后被重新投递");
            template.send("u48-t2", "acked").get(10, TimeUnit.SECONDS);
            out("kafkaManualAck redelivered=true atLeastOnce=true");

            // ---- 4. 错误处理：DefaultErrorHandler 按回退重试 3 次后经 recoverer 转投 DLT ----
            template.send("u48-t3", "poison").get(10, TimeUnit.SECONDS);
            check(ERROR_LATCH.await(20, TimeUnit.SECONDS), "错误处理最终应完成（转投 DLT）");
            check(ERROR_ATTEMPTS.get() == 3, "错误处理应投递 3 次，实际 " + ERROR_ATTEMPTS.get());
            out("kafkaErrorHandler attempts=3 recoveredTo=u48-t3-dlt");

            // ---- 5. 事务：提交的事务可见，中止的事务在 read_committed 下不可见 ----
            KafkaTemplate<String, String> txTemplate = context.getBean("txTemplate", KafkaTemplate.class);
            txTemplate.executeInTransaction(t -> {
                t.send("u48-t4", "tx-1");
                t.send("u48-t4", "tx-2");
                return true;
            });
            check(TX_LATCH.await(15, TimeUnit.SECONDS), "已提交事务的两条应被消费");
            try {
                txTemplate.executeInTransaction(t -> {
                    t.send("u48-t4", "tx-aborted");
                    throw new IllegalStateException("abort");
                });
            } catch (IllegalStateException expected) {
            }
            Thread.sleep(1000);
            int abortedVisible = (int) TX_CONSUMED.stream().filter(m -> m.contains("tx-aborted")).count();
            check(abortedVisible == 0, "中止事务在 read_committed 下不应可见");
            out("kafkaTransaction committedVisible=2 abortedInvisible=true");
        }
        out("cleanup contextClosed=true brokerManagedByRunner=true");
    }

    static void check(boolean condition, String message) {
        if (!condition) throw new AssertionError(message);
    }

    static void out(String line) {
        System.out.println("U48 " + line);
    }

    @Configuration
    @EnableKafka
    static class KafkaConfig {

        final String bootstrap = System.getProperty("u48.bootstrap", "127.0.0.1:29992");

        @Bean
        public KafkaAdmin kafkaAdmin() {
            Map<String, Object> configs = new HashMap<>();
            configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
            return new KafkaAdmin(configs);
        }

        @Bean
        public NewTopic t1() {
            return new NewTopic("u48-t1", 1, (short) 1);
        }

        @Bean
        public NewTopic t2() {
            return new NewTopic("u48-t2", 1, (short) 1);
        }

        @Bean
        public NewTopic t3() {
            return new NewTopic("u48-t3", 1, (short) 1);
        }

        @Bean
        public NewTopic t3dlt() {
            return new NewTopic("u48-t3-dlt", 1, (short) 1);
        }

        @Bean
        public NewTopic t4() {
            return new NewTopic("u48-t4", 1, (short) 1);
        }

        @Bean
        public ProducerFactory<String, String> producerFactory() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
            return new DefaultKafkaProducerFactory<>(props);
        }

        @Bean
        public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
            return new KafkaTemplate<>(producerFactory);
        }

        @Bean
        public DefaultKafkaProducerFactory<String, String> txProducerFactory() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
            props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "u48-tx-1");
            return new DefaultKafkaProducerFactory<>(props);
        }

        @Bean
        public KafkaTemplate<String, String> txTemplate(DefaultKafkaProducerFactory<String, String> txProducerFactory) {
            return new KafkaTemplate<>(txProducerFactory);
        }

        @Bean
        public KafkaTransactionManager<String, String> kafkaTransactionManager(
                DefaultKafkaProducerFactory<String, String> txProducerFactory) {
            return new KafkaTransactionManager<>(txProducerFactory);
        }

        @Bean
        public ConsumerFactory<String, String> consumerFactory() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, "u48-group");
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
            return new DefaultKafkaConsumerFactory<>(props);
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, String> batchFactory(
                ConsumerFactory<String, String> consumerFactory) {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory);
            return factory;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, String> manualFactory(
                ConsumerFactory<String, String> consumerFactory) {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory);
            factory.getContainerProperties().setAckMode(
                    org.springframework.kafka.listener.ContainerProperties.AckMode.MANUAL);
            return factory;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, String> errorFactory(
                ConsumerFactory<String, String> consumerFactory, KafkaTemplate<String, String> kafkaTemplate) {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory);
            DefaultErrorHandler handler = new DefaultErrorHandler(
                    new DeadLetterPublishingRecoverer(kafkaTemplate),
                    new org.springframework.util.backoff.FixedBackOff(10, 2));
            factory.setCommonErrorHandler(handler);
            return factory;
        }

        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, String> txFactory(
                ConsumerFactory<String, String> consumerFactory) {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            Map<String, Object> props = new HashMap<>(consumerFactory.getConfigurationProperties());
            props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
            factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
            return factory;
        }

        @KafkaListener(id = "mainListener", topics = "u48-t1", containerFactory = "batchFactory")
        public void listen(List<String> records) {
            records.forEach(r -> CONSUMED.add("t1:" + r));
            MAIN_LATCH.countDown();
        }

        @KafkaListener(id = "manualListener", topics = "u48-t2", containerFactory = "manualFactory")
        public void manual(String record, Acknowledgment acknowledgment) {
            CONSUMED.add("manual:" + record);
            MANUAL_LATCH.countDown();
            if ("acked".equals(record)) {
                acknowledgment.acknowledge();
            }
            // "no-ack" 不确认：重启后应重新投递
        }

        @KafkaListener(id = "errorListener", topics = "u48-t3", containerFactory = "errorFactory")
        public void failing(String record) {
            ERROR_ATTEMPTS.incrementAndGet();
            throw new IllegalStateException("always fails");
        }

        @KafkaListener(id = "dltListener", topics = "u48-t3-dlt", containerFactory = "batchFactory")
        public void dlt(String record) {
            ERROR_LATCH.countDown();
        }

        @KafkaListener(id = "txListener", topics = "u48-t4", containerFactory = "txFactory")
        public void tx(String record) {
            TX_CONSUMED.add("tx:" + record);
            TX_LATCH.countDown();
        }
    }
}