package org.example;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.clients.consumer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
public class ReportingKafkaConsumer {
    private static List<String> TOPICS = Arrays.asList("VALID_TOPIC", "SUSPICIOUS_TOPIC");
    private static final String BOOTSTRAP_SERVERS = "192.168.181.136:9092,localhost:9093,localhost:9094";
    static final Logger log = LoggerFactory.getLogger(ReportingKafkaConsumer.class);
    public static void main(String[] args) {
        String consumerGroup = "defaultAliConsumerGroup";
        if (args.length == 1) {
            consumerGroup = args[0];
        }
        System.out.println("Consumer is part of consumer group " + consumerGroup);
        Consumer<Long, TransactionInfo> kafkaConsumer = createKafkaConsumer(BOOTSTRAP_SERVERS, consumerGroup);
        System.out.println("created successfully");
        consumeMessages(TOPICS, kafkaConsumer);
    }
    public static void consumeMessages(List<String> topics, Consumer<Long, TransactionInfo> kafkaConsumer) {
        ObjectMapper objectMapper = new ObjectMapper();
        kafkaConsumer.subscribe(topics);
        System.out.println("subscribed successfully");
        while (true) {
            ConsumerRecords<Long, TransactionInfo> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            if (consumerRecords.isEmpty()) {
                continue;
            }
            for (ConsumerRecord<Long, TransactionInfo> record : consumerRecords) {
                TransactionInfo transactionInfo = record.value();
                String topic = record.topic();
                System.out.println("key : " + record.key());
                System.out.println("Received : " + transactionInfo.toString());
                try {
                    WritingHelper writingHelper = new WritingHelper(transactionInfo, topic);
                    System.out.println(writingHelper);
                    writingHelper.writeToLog();
                } catch (Exception e) {
                    System.out.println("Error Parsing");
                    System.out.println(e.getMessage());
                    log.error("Could Not Process Transaction: " + transactionInfo.toString());
                }
            }
            kafkaConsumer.commitAsync();
        }
    }
    public static Consumer<Long, TransactionInfo> createKafkaConsumer(String bootstrapServers, String consumerGroup) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, TransactionInfoDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        return new KafkaConsumer<>(properties);
    }
}
