/**
 * @file Application.java
 * @brief Implementation of the Transaction Reporting Service.
 */

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

import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * @class Application
 * @brief Centralized reporting service for all bank transactions.
 *
 * This service subscribes to both "valid-transactions" and "suspicious-transactions" 
 * topics. It acts as a logger that persists transaction history to the local 
 * disk, categorizing them into 'normal' and 'sus' folders for auditing and 
 * monthly statement generation.
 */
public class Application {

    /** @brief Kafka topic for cleared, non-fraudulent transactions. */
    private static final String VALID_TRANSACTIONS_TOPIC = "valid-transactions";
    
    /** @brief Kafka topic for flagged, potentially fraudulent transactions. */
    private static final String SUSPICIOUS_TRANSACTIONS_TOPIC = "suspicious-transactions";
    
    /** @brief Comma-separated list of Kafka broker addresses for the cluster. */
    private static final String BOOTSTRAP_SERVERS = "localhost:9092,localhost:9093";

    /**
     * @brief Application entry point.
     * 
     * Initializes the consumer and prepares an unmodifiable list of topics 
     * to subscribe to.
     * @param args Command line arguments.
     */
    public static void main(String[] args) {

        String consumerGroup = "transaction-reporting-service";

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

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

        // Subscribing to multiple topics for full visibility
        consumeMessages(
                Collections.unmodifiableList(
                        Arrays.asList(SUSPICIOUS_TRANSACTIONS_TOPIC, VALID_TRANSACTIONS_TOPIC)
                ),
                kafkaConsumer
        );
    }

    /**
     * @brief Consumes and processes records from multiple Kafka topics.
     * 
     * Subscribes to the provided list of topics and enters a polling loop. 
     * Each record is delegated to the reporting logic based on its source topic.
     * 
     * @param topics List of topic names to monitor.
     * @param kafkaConsumer The initialized Kafka consumer instance.
     */
    public static void consumeMessages(List<String> topics, Consumer<String, Transaction> kafkaConsumer) {

        kafkaConsumer.subscribe(topics);

        while (true) {
            ConsumerRecords<String, Transaction> records =
                    kafkaConsumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, Transaction> record : records) {
                Transaction transaction = record.value();
                
                /** @note Safeguard against deserialization errors. */
                if (transaction != null) {
                    recordTransactionForReporting(record.topic(), transaction);
                }
            }
        }
    }

    /**
     * @brief Configures a Kafka Consumer with shared reporting settings.
     * 
     * Enables auto-commit for reporting purposes, assuming a "best-effort" 
     * logging requirement where slight data loss on crash is acceptable 
     * for higher throughput.
     * 
     * @param bootstrapServers The Kafka cluster addresses.
     * @param consumerGroup The reporting service group ID.
     * @return Consumer<String, Transaction> A configured Kafka consumer.
     */
    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);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");

        return new KafkaConsumer<>(properties);
    }

    /**
     * @brief Branching logic to decide how to format and store reports.
     * 
     * Branches based on whether the message arrived via the 'suspicious' 
     * or 'valid' topic.
     * 
     * @param topic The topic the transaction was read from.
     * @param transaction The transaction data.
     */
    private static void recordTransactionForReporting(String topic, Transaction transaction) {

        String message;

        if (topic.equals(SUSPICIOUS_TRANSACTIONS_TOPIC)) {
            message = String.format(
                    "Suspicious transaction for user %s, amount $%.2f originating in %s",
                    transaction.getUser(),
                    transaction.getAmount(),
                    transaction.getTransactionLocation());

            System.out.println(message);
            saveToFile("../reports/sus", transaction.getUser(), message);

        } else if (topic.equals(VALID_TRANSACTIONS_TOPIC)) {
            message = String.format(
                    "Valid transaction for user %s, amount $%.2f",
                    transaction.getUser(),
                    transaction.getAmount());

            System.out.println(message);
            saveToFile("../reports/normal", transaction.getUser(), message);
        }
    }

    /**
     * @brief Utility method to persist reporting strings to the file system.
     * 
     * Resolves the path, creates directories if missing, and appends the 
     * message to a user-specific text file.
     * 
     * @param folderPath Destination directory (e.g., normal or sus).
     * @param user The username, used as the filename.
     * @param message The formatted report string.
     */
    private static void saveToFile(String folderPath, String user, String message) {
        try {
            Path dir = Paths.get(folderPath);

            if (!Files.exists(dir)) {
                Files.createDirectories(dir);
            }

            Path file = dir.resolve(user + ".txt");

            try (FileWriter writer = new FileWriter(file.toFile(), true)) {
                writer.write(message + System.lineSeparator());
            }

        } catch (IOException e) {
            System.err.println("Reporting file write failed: " + e.getMessage());
        }
    }
}