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.OnlineKontingentChangedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

@SuppressWarnings("unused")
public class KontingentSubscriber implements EventSubscriber<OnlineKontingentChangedEvent> {
    private static final Logger LOGGER = LoggerFactory.getLogger(KontingentSubscriber.class);

    private final FKKsVTInMemoryCache cache;

    private final EventNotification notification;

    public KontingentSubscriber(final EventBusFactory eventBusFactory,
                                final FKKsVTInMemoryCache cache,
                                final EventNotification notification) {
        EventBus<OnlineKontingentChangedEvent> kontingentBus = eventBusFactory.createOrGetEventBusFor(OnlineKontingentChangedEvent.class);
        kontingentBus.subscribe(OnlineKontingentChangedEvent.class, this, START_READING_FROM_BEGINNING);

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

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

        cache.addOrUpdate(event.onlineKontingent.vorfuehrungId(), getRawOrNull(event.onlineKontingent.neuesOnlineRestKontingent()));
        this.notification.notify(event);
    }

    public FKKsVTInMemoryCache getCache() {
        return cache;
    }
}
