package com.shapira.examples.streams.clickstreamenrich;

import com.google.gson.Gson;
import com.shapira.examples.streams.clickstreamenrich.model.PageView;
import com.shapira.examples.streams.clickstreamenrich.model.Search;
import com.shapira.examples.streams.clickstreamenrich.model.UserProfile;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * Ta klasa będzie generować symulowane kliknięcia, wyszukiwania i aktualizacje profilów
 * Dla uproszczenia będziemy generować tylko kilka zdarzeń - dwa profile, aktualizacja jednego profilu, trzy wyszukiwania, pięć kliknięć
 */
public class GenerateData {

    public static KafkaProducer<Integer, String> producer = null;

    public static void main(String[] args) throws Exception {

        System.out.println("Aby zatrzymać generowanie danych, naciśnij CTRL+C");

        List<ProducerRecord<Integer, String>> records = new ArrayList<>();
        Gson gson = new Gson();


        // Dodajemy zaczep zamykania
        Runtime.getRuntime().addShutdownHook(new Thread( () -> {
                System.out.println("Zamykanie");
                if (producer != null)
                    producer.close();
        }));



        // Dwóch użytkowników
        String[] interests1 = {"Surfowanie", "Wędrówki"};
        UserProfile user1 = new UserProfile(1, "Mateusz", "94301", interests1 );

        String[] interests2 = {"Narty", "Taniec"};
        UserProfile user2 = new UserProfile(2, "Anna", "94302", interests2);

        records.add(new ProducerRecord<>(Constants.USER_PROFILE_TOPIC, user1.getUserID(), gson.toJson(user1)));
        records.add(new ProducerRecord<>(Constants.USER_PROFILE_TOPIC, user2.getUserID(), gson.toJson(user2)));

        // Aktualizacja profilu

        String[] newInterests = {"Narty", "przetwarzanie strumieniowe"};

        records.add(new ProducerRecord<>(Constants.USER_PROFILE_TOPIC, user2.getUserID(), gson.toJson(user2.update("94303", newInterests))));


        // Dwa wyszukiwania

        Search search1 = new Search(1, "retro pianka do nurkowania");
        Search search2 = new Search(2, "lekka kurtka");


        records.add(new ProducerRecord<>(Constants.SEARCH_TOPIC, search1.getUserID(), gson.toJson(search1)));
        records.add(new ProducerRecord<>(Constants.SEARCH_TOPIC, search2.getUserID(), gson.toJson(search2)));

        // Trzy kliknięcia

        PageView view1 = new PageView(1, "collections/mens-wetsuits/products/w3-worlds-warmest-wetsuit");
        PageView view2 = new PageView(2, "product/womens-dirt-craft-bike-mountain-biking-jacket");
        PageView view3 = new PageView(2, "/product/womens-ultralight-down-jacket");

        records.add(new ProducerRecord<>(Constants.PAGE_VIEW_TOPIC, view1.getUserID(), gson.toJson(view1)));
        records.add(new ProducerRecord<>(Constants.PAGE_VIEW_TOPIC, view2.getUserID(), gson.toJson(view2)));
        records.add(new ProducerRecord<>(Constants.PAGE_VIEW_TOPIC, view3.getUserID(), gson.toJson(view3)));


        // Konfigurujemy producent
        // Jako klucza dla wszystkich zdarzeń użyjemy ID użytkownika, ponieważ łączenia wymagają wspólnego klucza
        // Ponieważ jako wartości będziemy zapisywać obiekty różnych typów, sami będziemy serializować je wszystkie do łańcuchów znaków JSON,
        // dlatego typ producer i serializator są tylko dla łańcuchów znaków



        Properties props = new Properties();

        props.put("bootstrap.servers", Constants.BROKER);
        props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // Uruchamiamy producent
        producer = new KafkaProducer<>(props);

        // Wysyłamy istniejące zdarzenia

        for (ProducerRecord<Integer, String> record: records)
            producer.send(record, (RecordMetadata r, Exception e) -> {
                if (e != null) {
                    System.out.println("Błąd podczas produkowania do tematu " + r.topic());
                    e.printStackTrace();
                }
            });


        // Uśpienie na pięć sekund, by mieć pewność, że nowe zdarzenia będą traktowane jako osobne sesje 
        records.clear();
        Thread.sleep(5000);

        // Jeszcze jedno wyszukiwanie

        Search search3 = new Search(2, "buty narciarskie z włókna węglowego");

        records.add(new ProducerRecord<>(Constants.SEARCH_TOPIC, search3.getUserID(), gson.toJson(search3)));

        // Dwa kliknięcia
        PageView view4 = new PageView(2, "product/salomon-quest-access-custom-heat-ski-boots-womens");
        PageView view5 = new PageView(2, "product/nordica-nxt-75-ski-boots-womens");

        records.add(new ProducerRecord<>(Constants.PAGE_VIEW_TOPIC, view4.getUserID(), gson.toJson(view4)));
        records.add(new ProducerRecord<>(Constants.PAGE_VIEW_TOPIC, view5.getUserID(), gson.toJson(view5)));

        // Kliknięcie dla nieznanego użytkownika bez wyszukiwań - chcemy mieć pewność, że dla tego też będziemy mieli wyniki

        PageView view6 = new PageView(-1, "product/osprey-atmos-65-ag-pack");
        records.add(new ProducerRecord<>(Constants.PAGE_VIEW_TOPIC, view6.getUserID(), gson.toJson(view6)));


        // Wysyłamy dodatkowe zdarzenia
        for (ProducerRecord<Integer, String> record: records)
            producer.send(record, (RecordMetadata r, Exception e) -> {
                if (e != null) {
                    System.out.println("Błąd podczas produkowania do tematu " + r.topic());
                    e.printStackTrace();
                }
            });


        // I kończymy...

        producer.close();


    }


}
