package com.github.msemys.esjc;


import com.github.msemys.esjc.event.ClientConnected;
import com.github.msemys.esjc.util.Strings;
import com.github.msemys.esjc.util.Subscriptions.DropData;
import com.github.msemys.esjc.util.concurrent.ResettableLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import static com.github.msemys.esjc.util.Numbers.isPositive;
import static com.github.msemys.esjc.util.Preconditions.checkArgument;
import static com.github.msemys.esjc.util.Preconditions.checkNotNull;
import static com.github.msemys.esjc.util.Ranges.BATCH_SIZE_RANGE;
import static com.github.msemys.esjc.util.Strings.defaultIfEmpty;
import static com.github.msemys.esjc.util.Strings.isNullOrEmpty;
import static com.github.msemys.esjc.util.Subscriptions.DROP_SUBSCRIPTION_EVENT;
import static com.github.msemys.esjc.util.Subscriptions.UNKNOWN_DROP_DATA;

/**
 * Catch-up subscription.
 */
public abstract class CatchUpSubscription implements AutoCloseable {
    protected final Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * The name of the stream to which the subscription is subscribed (empty if subscribed to $all stream).
     */
    public final String streamId;

    private final EventStore eventstore;
    private final boolean resolveLinkTos;
    private final UserCredentials userCredentials;
    protected final CatchUpSubscriptionListener listener;
    protected final int readBatchSize;
    protected final int maxPushQueueSize;
    private final Executor executor;

    private final Queue<ResolvedEvent> liveQueue = new ConcurrentLinkedQueue<>();
    private Subscription subscription;
    private final AtomicReference<DropData> dropData = new AtomicReference<>();
    private volatile boolean allowProcessing;
    private final AtomicBoolean isProcessing = new AtomicBoolean();
    protected volatile boolean shouldStop;
    private final AtomicBoolean isDropped = new AtomicBoolean();
    private final ResettableLatch stopped = new ResettableLatch(true);

    private final EventStoreListener reconnectionHook;

    protected CatchUpSubscription(EventStore eventstore,
                                  String streamId,
                                  boolean resolveLinkTos,
                                  CatchUpSubscriptionListener listener,
                                  UserCredentials userCredentials,
                                  int readBatchSize,
                                  int maxPushQueueSize,
                                  Executor executor) {
        checkNotNull(eventstore, "eventstore is null");
        checkNotNull(listener, "listener is null");
        checkNotNull(listener, "executor is null");
        checkArgument(BATCH_SIZE_RANGE.contains(readBatchSize), "readBatchSize is out of range. Allowed range: %s.", BATCH_SIZE_RANGE.toString());
        checkArgument(isPositive(maxPushQueueSize), "maxPushQueueSize should be positive");

        this.eventstore = eventstore;
        this.streamId = defaultIfEmpty(streamId, Strings.EMPTY);
        this.resolveLinkTos = resolveLinkTos;
        this.listener = listener;
        this.userCredentials = userCredentials;
        this.readBatchSize = readBatchSize;
        this.maxPushQueueSize = maxPushQueueSize;
        this.executor = executor;

        reconnectionHook = event -> {
            if (event instanceof ClientConnected) {
                onReconnect();
            }
        };
    }

    protected abstract void readEventsTill(EventStore eventstore,
                                           boolean resolveLinkTos,
                                           UserCredentials userCredentials,
                                           Long lastCommitPosition,
                                           Long lastEventNumber) throws Exception;

    protected abstract void tryProcess(ResolvedEvent event);

    void start() {
        logger.trace("Catch-up subscription to {}: starting...", streamId());
        runSubscription();
    }

    /**
     * Unsubscribes from the catch-up subscription.
     *
     * @param timeout the maximum wait time before it should timeout.
     * @throws TimeoutException when timeouts
     */
    public void stop(Duration timeout) throws TimeoutException {
        stop();
        logger.trace("Waiting on subscription to stop");
        if (!stopped.await(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
            throw new TimeoutException(String.format("Could not stop %s in time.", getClass().getSimpleName()));
        }
    }

    /**
     * Unsubscribes from the catch-up subscription asynchronously.
     */
    public void stop() {
        logger.trace("Catch-up subscription to {}: requesting stop...", streamId());

        logger.trace("Catch-up subscription to {}: unhooking from connection. Connected.", streamId());
        eventstore.removeListener(reconnectionHook);

        shouldStop = true;
        enqueueSubscriptionDropNotification(SubscriptionDropReason.UserInitiated, null);
    }

    /**
     * Unsubscribes from the catch-up subscription (using 2 seconds wait time before it should timeout).
     *
     * @throws TimeoutException when timeouts
     * @see #stop(Duration)
     */
    @Override
    public void close() throws TimeoutException {
        stop(Duration.ofSeconds(2));
    }

    private void onReconnect() {
        logger.trace("Catch-up subscription to {}: recovering after reconnection.", streamId());

        logger.trace("Catch-up subscription to {}: unhooking from connection. Connected.", streamId());
        eventstore.removeListener(reconnectionHook);

        while (dropData.get() != null && !isDropped.get()) {
            stopped.await(1, TimeUnit.SECONDS);
        }

        runSubscription();
    }

    private void runSubscription() {
        executor.execute(() -> {
            logger.trace("Catch-up subscription to {}: running...", streamId());

            stopped.reset();
            allowProcessing = false;
            isDropped.set(false);
            dropData.set(null);

            try {
                if (!shouldStop) {
                    logger.trace("Catch-up subscription to {}: pulling events...", streamId());
                    readEventsTill(eventstore, resolveLinkTos, userCredentials, null, null);
                }

                if (!shouldStop) {
                    logger.trace("Catch-up subscription to {}: subscribing...", streamId());

                    VolatileSubscriptionListener subscriptionListener = new VolatileSubscriptionListener() {
                        @Override
                        public void onEvent(Subscription s, ResolvedEvent event) {
                            if (dropData.get() == null) {
                                logger.trace("Catch-up subscription to {}: event appeared ({}, {}, {} @ {}).",
                                    streamId(), event.originalStreamId(), event.originalEventNumber(),
                                    event.originalEvent().eventType, event.originalPosition);

                                if (liveQueue.size() >= maxPushQueueSize) {
                                    enqueueSubscriptionDropNotification(SubscriptionDropReason.ProcessingQueueOverflow, null);
                                    subscription.unsubscribe();
                                } else {
                                    liveQueue.offer(event);
                                    if (allowProcessing) {
                                        ensureProcessingPushQueue();
                                    }
                                }
                            }
                        }

                        @Override
                        public void onClose(Subscription s, SubscriptionDropReason reason, Exception exception) {
                            enqueueSubscriptionDropNotification(reason, exception);
                        }
                    };

                    subscription = isSubscribedToAll() ?
                        eventstore.subscribeToAll(resolveLinkTos, subscriptionListener, userCredentials).get() :
                        eventstore.subscribeToStream(streamId, resolveLinkTos, subscriptionListener, userCredentials).get();

                    logger.trace("Catch-up subscription to {}: pulling events (if left)...", streamId());
                    readEventsTill(eventstore, resolveLinkTos, userCredentials, subscription.lastCommitPosition, subscription.lastEventNumber);
                }
            } catch (Exception e) {
                dropSubscription(SubscriptionDropReason.CatchUpError, e);
                return;
            }

            if (shouldStop) {
                dropSubscription(SubscriptionDropReason.UserInitiated, null);
                return;
            }

            logger.trace("Catch-up subscription to {}: processing live events...", streamId());
            listener.onLiveProcessingStarted(this);

            logger.trace("Catch-up subscription to {}: hooking to connection. Connected", streamId());
            eventstore.addListener(reconnectionHook);

            allowProcessing = true;
            ensureProcessingPushQueue();
        });
    }

    private void enqueueSubscriptionDropNotification(SubscriptionDropReason reason, Exception exception) {
        // if drop data was already set -- no need to enqueue drop again, somebody did that already
        if (dropData.compareAndSet(null, new DropData(reason, exception))) {
            liveQueue.offer(DROP_SUBSCRIPTION_EVENT);
            if (allowProcessing) {
                ensureProcessingPushQueue();
            }
        }
    }

    private void ensureProcessingPushQueue() {
        if (isProcessing.compareAndSet(false, true)) {
            executor.execute(this::processLiveQueue);
        }
    }

    private void processLiveQueue() {
        do {
            ResolvedEvent event;
            while ((event = liveQueue.poll()) != null) {
                // drop subscription artificial ResolvedEvent
                if (event.equals(DROP_SUBSCRIPTION_EVENT)) {
                    DropData previousDropData = dropData.getAndAccumulate(UNKNOWN_DROP_DATA,
                        (current, update) -> (current == null) ? update : current);

                    if (previousDropData == null) {
                        previousDropData = UNKNOWN_DROP_DATA;
                    }

                    dropSubscription(previousDropData.reason, previousDropData.exception);
                    isProcessing.compareAndSet(true, false);
                    return;
                }

                try {
                    tryProcess(event);
                } catch (Exception e) {
                    dropSubscription(SubscriptionDropReason.EventHandlerException, e);
                    return;
                }
            }
            isProcessing.compareAndSet(true, false);
        } while (!liveQueue.isEmpty() && isProcessing.compareAndSet(false, true));
    }

    private void dropSubscription(SubscriptionDropReason reason, Exception exception) {
        if (isDropped.compareAndSet(false, true)) {
            logger.trace("Catch-up subscription to {}: dropping subscription, reason: {}.", streamId(), reason, exception);

            if (subscription != null) {
                subscription.unsubscribe();
            }

            listener.onClose(this, reason, exception);

            stopped.release();
        }
    }

    /**
     * Determines whether or not this subscription is to $all stream or to a specific stream.
     *
     * @return {@code true} if this subscription is to $all stream, otherwise {@code false}
     */
    public boolean isSubscribedToAll() {
        return isNullOrEmpty(streamId);
    }

    /**
     * The last event number processed on the subscription.
     *
     * @return event number
     */
    public abstract long lastProcessedEventNumber();

    /**
     * The last position processed on the subscription.
     *
     * @return position
     */
    public abstract Position lastProcessedPosition();

    protected String streamId() {
        return defaultIfEmpty(streamId, "<all>");
    }

}
