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;

public class Application {

    private static final String VALID_TRANSACTIONS_TOPIC = "valid-transactions";
    private static final String SUSPICIOUS_TRANSACTIONS_TOPIC = "suspicious-transactions";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092,localhost:9093,localhost:9094";

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

        consumeMessages(
                Collections.unmodifiableList(
                        Arrays.asList(SUSPICIOUS_TRANSACTIONS_TOPIC, VALID_TRANSACTIONS_TOPIC)
                ),
                kafkaConsumer
        );
    }

    public static void consumeMessages(List<String> topics, Consumer<String, Transaction> kafkaConsumer) {

        // Subscribe to both topics
        kafkaConsumer.subscribe(topics);

        while (true) {

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

            for (ConsumerRecord<String, Transaction> record : records) {

                String topic = record.topic();
                Transaction transaction = record.value();

                recordTransactionForReporting(topic, transaction);
            }
        }
    }

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

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

    private static void saveToFile(String folderPath, String user, String message) {

        try {

            Path dir = Paths.get(folderPath);

            // create folder if it doesn't exist
            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) {
            e.printStackTrace();
        }
    }
}