/**
 * @file Application.java
 * @brief Implementation of the User Notification Service.
 */

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;

import org.apache.kafka.common.serialization.StringDeserializer;

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

/**
 * @class Application
 * @brief Consumer service responsible for alerting users of potential fraud.
 *
 * This microservice monitors the "suspicious-transactions" Kafka topic. When a 
 * suspicious transaction is detected by the Banking API, this service notifies 
 * the user via console output and logs the notification to a persistent file 
 * system for audit purposes.
 */
public class Application {

    /** @brief The Kafka topic containing transactions flagged as suspicious. */
    private static final String SUSPICIOUS_TRANSACTIONS_TOPIC = "suspicious-transactions";
    
    /** @brief Comma-separated list of Kafka broker addresses. */
    private static final String BOOTSTRAP_SERVERS = "localhost:9092,localhost:9093";

    /**
     * @brief Application entry point.
     * 
     * Sets up the consumer group and starts monitoring the suspicious transaction stream.
     * @param args Command line arguments.
     */
    public static void main(String[] args) {
        String consumerGroup = "user-notification-service";

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

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

        consumeMessages(SUSPICIOUS_TRANSACTIONS_TOPIC, kafkaConsumer);
    }

    /**
     * @brief Main processing loop for incoming suspicious transactions.
     * 
     * Continuously polls the Kafka topic and triggers user notifications 
     * for every received transaction.
     * 
     * @param topic The name of the suspicious transactions topic.
     * @param kafkaConsumer The initialized Kafka consumer instance.
     */
    public static void consumeMessages(String topic, Consumer<String, Transaction> kafkaConsumer) {

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

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

            for (ConsumerRecord<String, Transaction> record : records) {
                Transaction transaction = record.value();
                
                /** 
                 * @note Null check ensures the application doesn't crash 
                 * if a malformed message fails to deserialize.
                 */
                if (transaction != null) {
                    sendUserNotification(transaction);
                }
            }
        }
    }

    /**
     * @brief Configures and initializes a Kafka Consumer for notifications.
     * 
     * Sets the group ID and uses automatic offset committing for high throughput.
     * 
     * @param bootstrapServers The list of Kafka brokers.
     * @param consumerGroup The ID for the notification service group.
     * @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 Auto-commit is enabled here for simple "at-most-once" delivery semantics. */
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");

        return new KafkaConsumer<>(properties);
    }

    /**
     * @brief Notifies the user of a suspicious event and persists the alert to a file.
     * 
     * The method creates a directory called 'notifications' in the parent folder
     * and appends the alert message to a file named after the user.
     * 
     * @param transaction The suspicious transaction data.
     */
    private static void sendUserNotification(Transaction transaction) {

        String message = String.format(
                "Sending user %s notification about a suspicious transaction of $%.2f in their account originating in %s",
                transaction.getUser(),
                transaction.getAmount(),
                transaction.getTransactionLocation()
        );
    
        System.out.println(message);
    
        try {
            // Path configuration for notification storage
            Path notificationsDir = Paths.get("..", "notifications");
    
            // Ensure directory exists
            if (!Files.exists(notificationsDir)) {
                Files.createDirectories(notificationsDir);
            }
    
            // Define user-specific notification file
            String fileName = transaction.getUser() + ".txt";
            Path userFile = notificationsDir.resolve(fileName);
    
            /** @note File is opened in 'append' mode to keep a history of alerts. */
            try (FileWriter writer = new FileWriter(userFile.toFile(), true)) {
                writer.write(message + System.lineSeparator());
            }
    
        } catch (IOException e) {
            System.err.println("Failed to write user notification to file: " + e.getMessage());
        }
    }
}