/**
 * @file Application.java
 * @brief Main entry point for the Banking API Service.
 */

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @class Application
 * @brief The central orchestrator of the distributed banking system.
 *
 * The Banking API Service acts as the primary Producer. It receives incoming
 * credit card transactions, performs a fraud check by querying a Redis-backed
 * @ref UserResidenceDatabase, and routes the data to either the 'valid' or 
 * 'suspicious' Kafka topics.
 */
public class Application {
    
    /** @brief Topic name for transactions flagged as suspicious/fraudulent. */
    private static final String SUSPICIOUS_TOPIC = "suspicious-transactions";
    
    /** @brief Topic name for transactions cleared for processing. */
    private static final String VALID_TOPIC = "valid-transactions";
    
    /** @brief Comma-separated list of Kafka broker connection strings. */
    private static final String BOOTSTRAP_SERVERS = "localhost:9092,localhost:9093";

    /**
     * @brief Main method to start the Banking API Service.
     * 
     * Initializes the Kafka producer and begins processing the transaction stream.
     * Ensures the producer is flushed and closed gracefully on shutdown.
     * 
     * @param args Command line arguments.
     */
    public static void main(String[] args) {
        Producer<String, Transaction> kafkaProducer = createKafkaProducer(BOOTSTRAP_SERVERS);
        try {
            processTransactions(new IncomingTransactionsReader(), new UserResidenceDatabase(), kafkaProducer);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // Cleanly close resources
            kafkaProducer.flush();
            kafkaProducer.close();
        }
    }

    /**
     * @brief Orchestrates the flow from transaction ingestion to Kafka production.
     * 
     * This method implements the core fraud detection logic:
     * 1. Retrieve the user's home country from Redis.
     * 2. Compare it to the transaction's reported location.
     * 3. Route the message to the appropriate topic using the user ID as the Kafka Key.
     * 
     * @details Robustness: If a user is missing from the database (e.g., 'Sara'), 
     * the method catches the @ref RuntimeException, logs an error, and continues 
     * processing the next transaction to prevent service downtime.
     * 
     * @param reader The source providing incoming transaction objects.
     * @param db The Redis-based residence database for fraud checks.
     * @param producer The Kafka producer used for message routing.
     */
    public static void processTransactions(IncomingTransactionsReader reader, 
                                           UserResidenceDatabase db, 
                                           Producer<String, Transaction> producer) {
        while (reader.hasNext()) {
            Transaction transaction = reader.next();
            try {
                // Step 1: Data Enrichment via Redis
                String residence = db.getUserResidence(transaction.getUser());

                // Step 2: Fraud Detection Decision
                String topic = transaction.getTransactionLocation().equals(residence) ? VALID_TOPIC : SUSPICIOUS_TOPIC;
                
                /** 
                 * @note Key-Based Routing: Passing transaction.getUser() as the Key 
                 * ensures all transactions for the same user land in the same 
                 * Kafka partition, guaranteeing processing order.
                 */
                ProducerRecord<String, Transaction> record = new ProducerRecord<>(topic, transaction.getUser(), transaction);
                producer.send(record);

                System.out.println("Successfully processed transaction for: " + transaction.getUser());

            } catch (RuntimeException e) {
                // Skip malformed or unknown user data
                System.err.println("SKIPPING TRANSACTION: " + e.getMessage());
            }
        }
    }

    /**
     * @brief Configures and initializes a high-reliability Kafka Producer.
     * 
     * Applies settings for Exactly-Once Semantics (EOS) to prevent double-charging 
     * or duplicate records in the event of network retries.
     * 
     * @param bootstrapServers The addresses of the Kafka cluster.
     * @return Producer<String, Transaction> A configured Kafka producer instance.
     */
    public static Producer<String, Transaction> createKafkaProducer(String bootstrapServers) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "banking-api-service");

        /** 
         * @details Exactly-Once Semantics (EOS):
         * - Idempotence: Assigns sequence numbers to prevent duplicates.
         * - Acks=all: Ensures all replicas acknowledge the write.
         */
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); 
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        // Serializers for Key (String) and Value (Transaction)
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Transaction.TransactionSerializer.class.getName());

        return new KafkaProducer<>(props);
    }
}