/**
 * @file Application.java
 * @brief Implementation of the Account Manager Service.
 */

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

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

/**
 * @class Application
 * @brief Consumer service that manages and authorizes valid financial transactions.
 *
 * The Account Manager service subscribes to the "valid-transactions" Kafka topic.
 * Its primary responsibility is to simulate the final authorization of funds
 * and ensure that transactions are processed reliably using manual offset commits.
 */
public class Application {
    
    /** @brief The Kafka topic containing transactions cleared by the Banking API. */
    private static final String VALID_TRANSACTIONS_TOPIC = "valid-transactions";
    
    /** @brief Comma-separated list of Kafka broker addresses. */
    private static final String BOOTSTRAP_SERVERS = "localhost:9092,localhost:9093";

    /**
     * @brief Application entry point.
     * 
     * Initializes the consumer group and starts the infinite message consumption loop.
     * @param args Command line arguments (not used).
     */
    public static void main(String[] args) {
        String consumerGroup = "account-manager";

        System.out.println("Consumer is part of consumer group " + consumerGroup);

        Consumer<String, Transaction> kafkaConsumer = createKafkaConsumer(BOOTSTRAP_SERVERS, consumerGroup);

        consumeMessages(VALID_TRANSACTIONS_TOPIC, kafkaConsumer);
    }

    /**
     * @brief Subscription and processing loop for Kafka messages.
     * 
     * This method subscribes to the specified topic, polls Kafka for new records,
     * processes them by calling @ref approveTransaction, and manually commits 
     * offsets to ensure data consistency.
     * 
     * @param topic The name of the topic to subscribe to.
     * @param kafkaConsumer The initialized Kafka consumer instance.
     */
    public static void consumeMessages(String topic, Consumer<String, Transaction> kafkaConsumer) {
        kafkaConsumer.subscribe(Collections.singletonList(topic));

        while (true) {
            // Poll for a batch of records
            ConsumerRecords<String, Transaction> records = kafkaConsumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, Transaction> record : records) {
                Transaction transaction = record.value();

                if (transaction != null) {
                    approveTransaction(transaction);
                }
            }

            /** 
             * @note Manual offset management: We commit offsets only after 
             * the batch has been successfully processed to prevent data loss.
             */
            kafkaConsumer.commitAsync();
        }
    }

    /**
     * @brief Configures and initializes a Kafka Consumer.
     * 
     * Sets up deserializers for @ref Transaction objects and disables auto-commit 
     * to allow for manual offset management.
     * 
     * @param bootstrapServers The list of Kafka brokers.
     * @param consumerGroup The ID of the consumer group this instance belongs to.
     * @return Consumer<String, Transaction> A configured Kafka consumer instance.
     */
    public static Consumer<String, Transaction> createKafkaConsumer(String bootstrapServers, String consumerGroup) {
        Properties properties = new Properties();

        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Transaction.TransactionDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
        
        /** 
         * @details Manual offset management is enabled by setting auto-commit to false.
         * 'earliest' ensures the consumer reads from the start of the topic if no offset exists.
         */
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        return new KafkaConsumer<>(properties);
    }

    /**
     * @brief Simulates the authorization and fund transfer process.
     * 
     * In a real system, this would interact with a core banking ledger to 
     * move money from the user's account to the merchant.
     * 
     * @param transaction The valid transaction to be approved.
     */
    private static void approveTransaction(Transaction transaction) {
        System.out.println("----------------------------------------------------------------");
        System.out.println(String.format("Authorizing transaction for user: %s", transaction.getUser()));
        System.out.println(String.format("Action: Transferring $%.2f from User Account to the Merchant at %s", 
                transaction.getAmount(), 
                transaction.getTransactionLocation()));
        System.out.println("Status: SUCCESS - Funds Dispatched.");
        System.out.println("----------------------------------------------------------------");
    }
}