package de.accso.flexinale.besucherportal.api_in.event;

import de.accso.flexinale.backoffice.api_contract.event.VorfuehrungCreatedEvent;
import de.accso.flexinale.backoffice.api_contract.event.VorfuehrungDeletedEvent;
import de.accso.flexinale.backoffice.api_contract.event.VorfuehrungUpdatedEvent;
import de.accso.flexinale.besucherportal.application.services.FKKsVTInMemoryCache;
import de.accso.flexinale.common.api.eventbus.EventBus;
import de.accso.flexinale.common.api.eventbus.EventBusFactory;
import de.accso.flexinale.common.api.eventbus.EventNotification;
import de.accso.flexinale.common.api.eventbus.EventSubscriber;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static de.accso.flexinale.common.api.eventbus.EventSubscriptionAtStart.START_READING_FROM_BEGINNING;

@SuppressWarnings("unused")
public class VorfuehrungSubscriber implements EventSubscriber.EventSubscriber3<VorfuehrungCreatedEvent, VorfuehrungUpdatedEvent, VorfuehrungDeletedEvent> {
    private static final Logger LOGGER = LoggerFactory.getLogger(VorfuehrungSubscriber.class);

    private final FKKsVTInMemoryCache cache;

    private final EventNotification notification;

    @SuppressWarnings({"RedundantCast", "unchecked", "rawtypes"})
    public VorfuehrungSubscriber(final EventBusFactory eventBusFactory,
                                 final FKKsVTInMemoryCache cache,
                                 final EventNotification notification) {
        EventBus<VorfuehrungCreatedEvent> createBus = eventBusFactory.createOrGetEventBusFor(VorfuehrungCreatedEvent.class);
        EventBus<VorfuehrungUpdatedEvent> updateBus = eventBusFactory.createOrGetEventBusFor(VorfuehrungUpdatedEvent.class);
        EventBus<VorfuehrungDeletedEvent> deleteBus = eventBusFactory.createOrGetEventBusFor(VorfuehrungDeletedEvent.class);

        createBus.subscribe(VorfuehrungCreatedEvent.class, (EventSubscriber) this, START_READING_FROM_BEGINNING);
        updateBus.subscribe(VorfuehrungUpdatedEvent.class, (EventSubscriber) this, START_READING_FROM_BEGINNING);
        deleteBus.subscribe(VorfuehrungDeletedEvent.class, (EventSubscriber) this, START_READING_FROM_BEGINNING);

        this.cache = cache;
        this.notification = notification;
    }

    @Override
    public String getName() {
        return VorfuehrungSubscriber.class.getName();
    }

    @Override
    public String getGroupName() {
        return "flexinale-distributed-besucherportal"; //TODO make configurable, use "spring.kafka.consumer.group-id" from application.properties
    }

    @Override
    public void receive(final VorfuehrungCreatedEvent event) {
        LOGGER.debug("received new event " + event);

        cache.addOrUpdate(event.vorfuehrung);
        this.notification.notify(event);
    }

    @Override
    public void receive2(final VorfuehrungUpdatedEvent event) {
        LOGGER.debug("received new event " + event);

        cache.addOrUpdate(event.vorfuehrung);
        this.notification.notify(event);
    }

    @Override
    public void receive3(final VorfuehrungDeletedEvent event) {
        LOGGER.debug("received new event " + event);

        cache.delete(event.vorfuehrung);
        this.notification.notify(event);
    }

    public FKKsVTInMemoryCache getCache() {
        return cache;
    }
}
