package org.example;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class AccountManagerKafkaConsumer {
    private static final String TOPIC = "VALID_TOPIC";
    private static final String BOOTSTRAP_SERVERS = "192.168.181.136:9092,localhost:9093,localhost:9094";

    public static final Logger log = LoggerFactory.getLogger(AccountManagerKafkaConsumer.class);

    public static void main(String[] args) {
        String consumerGroup = "defaultConsumerGroup";//"ValidTransactionsGroup";
        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);

        consumeMessages(TOPIC, kafkaConsumer);
    }

    public static void consumeMessages(String topic, Consumer<Long, TransactionInfo> kafkaConsumer) {
        ObjectMapper objectMapper = new ObjectMapper();

        kafkaConsumer.subscribe(Collections.singletonList(topic));

        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();
                System.out.println("Key: " + record.key());
                System.out.println("Received: " + transactionInfo.toString());
                try {
                    WritingHelper writingHelper = new WritingHelper(transactionInfo);
                    writingHelper.writeToLog();
                    boolean success = DataBaseHandler.editAmount(transactionInfo.getName(), transactionInfo.getAmount());
                    if (!success) {
                        log.warn(transactionInfo.getName() + ": Insufficient balance to withdraw amount");
                    }
                    System.out.println(writingHelper);
                } catch (Exception e) {
                    System.out.println("Error processing transaction");
                    System.out.println(e.getMessage());
                    log.error("Failed to process transaction: " + transactionInfo.toString(), e);
                }
            }

            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);
    }
}