import {KafkaClient, Producer} from "kafka-node";
import EventStore from "../application/EventStore";
import config from "../config";
import ActionInteraction from "../domain/ActionInteraction";
import AnswerInteraction from "../domain/AnswerInteraction";
import CommentInteraction from "../domain/CommentInteraction";
import Exploration from "../domain/Exploration";

const client: KafkaClient = new KafkaClient({
    kafkaHost: config.kafka_server,
});

const  producer: Producer = new Producer(client);

export default class EventStoreKafka implements EventStore {

    public notifySessionExploration(sessionId: string, exploration: Exploration): Promise<void> {
        const sequence: Array<{
            concreteType: string,
            kind: string,
            value: string,
        }> = exploration.interactionList.map((interaction) => {
            if (interaction instanceof ActionInteraction) {
                return {
                    concreteType : "Action",
                    kind : interaction.action.kind,
                    value : interaction.action.value,
                };
            }
            if (interaction instanceof CommentInteraction) {
                return {
                    concreteType : "Comment",
                    kind : interaction.comment.kind,
                    value : interaction.comment.value,
                };
            }
            if (interaction instanceof AnswerInteraction) {
                return {
                    concreteType : "Answer",
                    kind : interaction.answer.kind,
                    value : interaction.answer.value,
                };
            }
        });

        return new Promise ((resolve, reject) => {
            producer.send(
                [{
                messages: JSON.stringify({
                    exploration: sequence,
                    kind: "exploration",
                    sessionId,
                    explorationNumber: exploration.explorationNumber,
                    testerName: exploration.tester.name,
                }),
                topic: "MAIN_TOPIC",
            }], (err) => {
                if (err) {
                    console.error("Error in EventStoreKafkaDelegate", err);
                    reject(err);
                } else {
                    resolve();
                }
            });
        });
    }
}
