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

import de.accso.flexinale.backoffice.api_contract.event.VorfuehrungCreatedEvent;
import de.accso.flexinale.backoffice.api_contract.event.VorfuehrungUpdatedEvent;
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.common.shared_kernel.FlexinaleIllegalStateException;
import de.accso.flexinale.ticketing.application.services.TicketService;
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 VorfuehrungSubscriber implements EventSubscriber.EventSubscriber2<VorfuehrungCreatedEvent, VorfuehrungUpdatedEvent> {
    private static final Logger LOGGER = LoggerFactory.getLogger(VorfuehrungSubscriber.class);

    final TicketService ticketService;

    private final EventNotification notification;

    @SuppressWarnings({"RedundantCast", "unchecked"})
    public VorfuehrungSubscriber(final EventBusFactory eventBusFactory,
                                 final TicketService ticketService,
                                 final EventNotification notification) {
        EventBus<VorfuehrungCreatedEvent> eventBusCreated = eventBusFactory.createOrGetEventBusFor(VorfuehrungCreatedEvent.class);
        EventBus<VorfuehrungUpdatedEvent> eventBusUpdated = eventBusFactory.createOrGetEventBusFor(VorfuehrungUpdatedEvent.class);

        eventBusCreated.subscribe(VorfuehrungCreatedEvent.class, (EventSubscriber) this, START_READING_FROM_BEGINNING);
        eventBusUpdated.subscribe(VorfuehrungUpdatedEvent.class, (EventSubscriber) this, START_READING_FROM_BEGINNING);

        this.ticketService = ticketService;
        this.notification = notification;
    }

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

    @Override
    public String getGroupName() {
        return "flexinale-distributed-ticketing"; //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);

        this.notification.notify(event);

        try {
            // check if Vorfuehrung is already there (then do nothing, event was sent more than once)
            if ( ! ticketService.vorfuehrungIstBereitsImVerkauf(event.vorfuehrung.id())) {
                ticketService.gebeVorfuehrungInDenVerkauf(event.vorfuehrung.id(), getRawOrNull(event.vorfuehrung.kinoSaal().anzahlPlaetze()));
            }
        }
        catch (FlexinaleIllegalStateException ex) {
            LOGGER.warn("received VorfuehrungCreatedEvent ignored as Kontingent already exist for Vorfuehrung %s"
                    .formatted(event.vorfuehrung.id()), ex);
        }
    }

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

        this.notification.notify(event);

        if ( ! ticketService.vorfuehrungIstBereitsImVerkauf(event.vorfuehrung.id())){
            // Events might not be delivered in same order as sent: So the Updated event might have overtaken a Created event.
            LOGGER.warn("received VorfuehrungUpdatedEvent although Vorfuehrung does not yet exist: %s Handling this as VorfuehrungCreated"
                    .formatted(event.vorfuehrung.id()));
            ticketService.gebeVorfuehrungInDenVerkauf(event.vorfuehrung.id(), getRawOrNull(event.vorfuehrung.kinoSaal().anzahlPlaetze()));
        }
        ticketService.aktualisiereVorfuehrung(event.vorfuehrung.id(), getRawOrNull(event.vorfuehrung.kinoSaal().anzahlPlaetze()));
    }
}
