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

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 de.accso.flexinale.ticketing.api_contract.event.TicketGekauftEvent;
import de.accso.flexinale.ticketing.api_contract.event.TicketUngueltigEvent;
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 TicketSubscriber implements EventSubscriber.EventSubscriber2<TicketGekauftEvent, TicketUngueltigEvent> {
    private static final Logger LOGGER = LoggerFactory.getLogger(TicketSubscriber.class);

    private final FKKsVTInMemoryCache cache;

    private final EventNotification notification;

    @SuppressWarnings({"RedundantCast", "unchecked", "rawtypes"})
    public TicketSubscriber(final EventBusFactory eventBusFactory,
                            final FKKsVTInMemoryCache cache,
                            final EventNotification notification) {
        EventBus<TicketGekauftEvent> createBus = eventBusFactory.createOrGetEventBusFor(TicketGekauftEvent.class);
        EventBus<TicketUngueltigEvent> ungueltigBus = eventBusFactory.createOrGetEventBusFor(TicketUngueltigEvent.class);

           createBus.subscribe(TicketGekauftEvent.class,   (EventSubscriber) this, START_READING_FROM_BEGINNING);
        ungueltigBus.subscribe(TicketUngueltigEvent.class, (EventSubscriber) this, START_READING_FROM_BEGINNING);

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

    @Override
    public String getName() {
        return TicketSubscriber.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 TicketGekauftEvent event) {
        LOGGER.debug("received new event " + event);

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

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

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

    public FKKsVTInMemoryCache getCache() {
        return cache;
    }
}
