package com.atlassian.bitbucket.mesh.repair;

import com.atlassian.bitbucket.mesh.git.exception.AbortedException;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.Closeable;
import java.util.Comparator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.BiPredicate;

import static java.util.Objects.requireNonNull;

/**
 * Utility class that efficiently compares two sorted {@link Sequence sequences} of items, called '{@link #from() from}'
 * and '{@link #to() to}'. Items are provided to either sequence through {@link Sequence#onNext(Object)} and the end
 * of the sequence is signalled by a call to {@link Sequence#close()}.
 * <p>
 * Detected differences are emitted to the provided {@link DiffCallback callback}.
 *
 * @param <T> the item type
 */
class Differ<T> {

    private static final Object CLOSED = new Object();

    private final DiffCallback<T> callback;
    private final Comparator<T> comparator;
    private final BiPredicate<T, T> equalityPredicate;
    private final SimpleSequence from;
    private final Object lock;
    private final SimpleSequence to;
    private final CompletableFuture<Void> result;

    /**
     * @param comparator        comparator representing the sort order of the to and from item sequences
     * @param equalityPredicate used to determine whether two items that are the same item based on sorting, have the
     *                          same value (e.g. whether two refs with the same ref ID reference the same object)
     * @param callback          callback to invoke for a detected difference between from and to items
     */
    Differ(@Nonnull Comparator<T> comparator, @Nonnull BiPredicate<T, T> equalityPredicate,
           @Nonnull DiffCallback<T> callback) {
        this.callback = callback;
        this.comparator = comparator;
        this.equalityPredicate = equalityPredicate;

        from = new SimpleSequence();
        lock = new Object();
        result = new CompletableFuture<>();
        to = new SimpleSequence();
    }

    @Nonnull
    Sequence<T> from() {
        return from;
    }

    @Nonnull
    Sequence<T> to() {
        return to;
    }

    @SuppressWarnings("unchecked")
    private void onUpdated() {
        while (true) {
            synchronized (lock) {
                if (result.isDone()) {
                    result.join();
                    return;
                }

                Object fromObject = from.peek();
                Object toObject = to.peek();
                if (fromObject == CLOSED && toObject == CLOSED) {
                    result.complete(null);
                    return;
                }
                if (fromObject == null || toObject == null) {
                    //Not enough information to diff items; more items need to be provided through
                    //the from and/or to sequences before we can compare the next items.
                    return;
                }
                if (fromObject == CLOSED) {
                    //To has a item that from does not have
                    callback.onDifference(null, (T) toObject);
                    to.pop();
                } else if (toObject == CLOSED) {
                    //From has an item that to does not have
                    callback.onDifference((T) fromObject, null);
                    from.pop();
                } else {
                    //Both from and to have an item
                    T fromItem = (T) fromObject;
                    T toItem = (T) toObject;
                    int compareResult = comparator.compare(fromItem, toItem);
                    if (compareResult == 0) {
                        //Both sides have the same item
                        if (!equalityPredicate.test(fromItem, toItem)) {
                            //But their values differ
                            callback.onDifference(fromItem, toItem);
                        }
                        from.pop();
                        to.pop();
                    } else if (compareResult < 0) {
                        //From has an item that to does not have
                        callback.onDifference(fromItem, null);
                        from.pop();
                    } else {
                        //To has an item that from does not have
                        callback.onDifference(null, toItem);
                        to.pop();
                    }
                }
            }
        }
    }

    private void handleInterrupted(InterruptedException e) {
        result.completeExceptionally(new AbortedException("Interrupted", e));
        //Restore the interrupted flag
        Thread.currentThread().interrupt();
    }

    @FunctionalInterface
    interface DiffCallback<T> {

        void onDifference(@Nullable T from, @Nullable T to);
    }

    interface Sequence<T> extends Closeable {

        /**
         * Called to signal the end of the sequence
         */
        @Override
        void close();

        /**
         * @param item adds an item to the sequence
         */
        void onNext(@Nonnull T item);
    }

    private class SimpleSequence implements Sequence<T> {

        private final BlockingQueue<Object> items;

        SimpleSequence() {
            items = new LinkedBlockingQueue<>(5);
        }

        @Override
        public void close() {
            try {
                items.put(CLOSED);
                onUpdated();
            } catch (InterruptedException e) {
                handleInterrupted(e);
            }
        }

        @Override
        public void onNext(@Nonnull T item) {
            try {
                items.put(requireNonNull(item, "item"));
                onUpdated();
            } catch (InterruptedException e) {
                handleInterrupted(e);
            }
        }

        private Object peek() {
            return items.peek();
        }

        private void pop() {
            try {
                items.take();
            } catch (InterruptedException e) {
                handleInterrupted(e);
            }
        }
    }
}
