package com.example.support;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.UUID;

/**
 * Owned event/async test client for Apache Kafka (L3git). Produce a record to a topic, or consume
 * from a topic and assert a matching record arrives within a timeout - so you can test event-driven
 * flows end to end. Thin wrapper over the standard {@code kafka-clients} (String key/value). The
 * kafka-clients dependency is added to the project's pom only when a test uses a Kafka step, so
 * projects that don't need it stay lean. Generated / shipped by L3git.
 */
public class KafkaClient {

    public void produce(String bootstrap, String topic, String key, String value) {
        Properties p = new Properties();
        p.put("bootstrap.servers", bootstrap);
        p.put("key.serializer", StringSerializer.class.getName());
        p.put("value.serializer", StringSerializer.class.getName());
        p.put("acks", "all");
        try (Producer<String, String> producer = new KafkaProducer<>(p)) {
            producer.send(new ProducerRecord<>(topic, key, value));
            producer.flush();
        }
    }

    /** Consume from {@code topic} until a record value contains {@code contains} (blank = any) or timeout. */
    public String consumeExpect(String bootstrap, String topic, String contains, long timeoutMs) {
        Properties p = new Properties();
        p.put("bootstrap.servers", bootstrap);
        p.put("group.id", "l3git-" + UUID.randomUUID());
        p.put("key.deserializer", StringDeserializer.class.getName());
        p.put("value.deserializer", StringDeserializer.class.getName());
        p.put("auto.offset.reset", "earliest");
        p.put("enable.auto.commit", "false");
        try (Consumer<String, String> consumer = new KafkaConsumer<>(p)) {
            consumer.subscribe(Collections.singletonList(topic));
            long end = System.currentTimeMillis() + timeoutMs;
            while (System.currentTimeMillis() < end) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                for (ConsumerRecord<String, String> r : records) {
                    String v = String.valueOf(r.value());
                    if (contains == null || contains.isEmpty() || v.contains(contains)) { return v; }
                }
            }
        }
        throw new AssertionError("Kafka: no record on topic \"" + topic + "\""
                + (contains == null || contains.isEmpty() ? "" : " containing \"" + contains + "\"")
                + " within " + timeoutMs + "ms");
    }
}
