package com.atlassian.bitbucket.util;

import com.google.common.collect.Iterables;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.*;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Stream;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.util.Objects.requireNonNull;

public class PageUtils {

    protected PageUtils() {
        throw new UnsupportedOperationException(getClass().getName() +
                " is a utility class and should not be instantiated");
    }

    /**
     * Safely cast {@code page} to a page of type {@code <S>}, where {@code S} is a super type of {@code ST}.
     * <p>
     * This method is for convenient casting of {@link Page pages} holding values of particular subtype {@code S}
     * to a chosen supertype {@code ST} of {@code S}. For instance, given a {@code Page<Integer>}, this method can be
     * used to conveniently obtain a reference to the same page instance as a {@code Page<Number>}:
     * <pre>
     * {@code
     *     Page<Integer> ints = // ...
     *     Page<Number> numbers = PageUtils.asPageOf(Number.class, ints);
     * }
     * </pre>
     * <p>
     * NOTE: this method enforces compile-type safety only and will not raise an error if the supplied page contains
     * values that are not assignable to the requested resulting type. As an example, if the {@code Page<Number>}
     * contained {@code String} values (e.g. as a result of unchecked assignments), this method will still return
     * a reference to {@link Page} of requested component type {@code Number}, even though the values are not
     * compatible with it.
     * <p>
     * This method does not iterate over the values of the supplied {@code page} or modify it in any other way, and it
     * <i>always</i> returns a reference to the supplied {@code page} instance.
     *
     * @param supertype super type to coerce page component type to; may not be {@literal null}
     * @param page      page reference to coerce, can be {@literal null}, in which case {@literal null} will be returned
     * @param <S>       super type parameter
     * @param <ST>      subtype parameter
     * @return the same {@code page} instance, as a page of type {@code <S>}, {@literal null} will be returned if
     *         {@code page} was {@literal null}.
     */
    @SuppressWarnings("unchecked")
    public static <S, ST extends S, P extends Page<S>, SP extends Page<ST>> P asPageOf(@Nonnull Class<S> supertype,
                                                                                       @Nullable SP page) {
        checkNotNull(supertype, "supertype");
        return (P) page;
    }

    public static <T> Page<T> createEmptyPage(PageRequest pageRequest) {
        return createPage(Collections.<T>emptyList(), pageRequest);
    }

    public static <T> Page<T> createPage(Iterable<T> elements, PageRequest request) {
        int size = Iterables.size(elements);
        boolean lastPage = size <= request.getLimit();
        int pageSize = lastPage ? size : request.getLimit();

        return new PageImpl<>(request, pageSize, Iterables.limit(elements, pageSize), lastPage);
    }

    public static <T> Page<T> createPage(Iterable<T> elements, boolean lastPage, PageRequest request) {
        int size = Iterables.size(elements);
        int pageSize = lastPage ? size : request.getLimit();

        return new PageImpl<>(request, pageSize, Iterables.limit(elements, pageSize), lastPage);
    }

    public static <T> Page<T> createPage(PageRequest request, Function<Integer, T> elementGenerator) {
        List<T> values = new ArrayList<>();
        int num = request.getLimit() + 1;
        for (int i = 0; i < num; ++i) {
            int index = request.getStart() + i;
            T element = elementGenerator.apply(index);
            if (element != null) {
                values.add(element);
            } else {
                break;
            }
        }
        return createPage(values, request);
    }

    /**
     * Retrieves unfiltered pages using the {@link PageProvider} and filters them with the provided {@code Predicate}.
     *
     * @param provider  retrieves unfiltered pages
     * @param predicate used to filter pages retrieved from the {@link PageProvider}
     * @param request   defines the start and limit of the page to be returned
     * @param <T> the value type for the returned page
     * @return a filtered page containing zero or more values which match the {@code Predicate}
     */
    @Nonnull
    public static <T> Page<T> filterPages(PageProvider<T> provider, Predicate<? super T> predicate,
                                          PageRequest request) {
        int internalStart = request.getStart();
        //Internal fetch size is twice the size requested to increase the likelihood of getting a full page
        int internalLimit = request.getLimit() * 2;
        //To determine whether there's another page, we need to find at least 1 more match than they asked for
        int pageLimit = request.getLimit() + 1;

        SortedMap<Integer, T> results = new TreeMap<>();

        Page<T> internalPage;
        outer: do {
            PageRequest internalRequest = new PageRequestImpl(internalStart, internalLimit);
            internalPage = provider.get(internalRequest);

            for (Map.Entry<Integer, T> entry : internalPage.getOrdinalIndexedValues().entrySet()) {
                if (predicate.test(entry.getValue())) {
                    results.put(entry.getKey(), entry.getValue());

                    if (results.size() == pageLimit) {
                        //Once we have a full page (which is really a full page plus one), break out of the
                        //*entire* loop, not just the for loop.
                        break outer;
                    }
                }
            }
            internalStart += internalLimit;
        } while (results.size() < pageLimit && !internalPage.getIsLastPage());

        int size = results.size();
        boolean lastPage = size < pageLimit;
        int pageSize = lastPage ? size : request.getLimit();

        return new FilteredPageImpl<>(request, pageSize, results, lastPage);
    }

    /**
     * A utility method to create a new page request without having to explicitly choose an implementation class.
     *
     * @param start the start of the page request
     * @param limit the limit of the page request
     * @return the non-null page request
     */
    @Nonnull
    public static PageRequest newRequest(int start, int limit) {
        return new PageRequestImpl(start, limit);
    }

    /**
     * A utility method to reduce the size of a page.
     *
     * @param page the original page
     * @param limit the new limit of the page (must be lower than the original page's limit)
     * @param request the original page request
     * @return the non-null page request
     */
    @Nonnull
    public static <T> Page<T> reducePage(Page<T> page, int limit, PageRequest request) {
        checkArgument(request.getLimit() >= limit);
        return new PageImpl<>(request.buildRestrictedPageRequest(limit),
                Iterables.limit(page.getValues(), limit), page.getSize() <= limit);
    }

    /**
     * A utility method to convert a {@link PageProvider} of {@code T} to a {@link Iterable} of {@code T}
     *
     * @param pageProvider a page provider
     * @param batchSize    the size of the page to fetch
     * @return a {@link Iterable} of objects of type {@code T}
     * @since 4.8
     */
    @Nonnull
    public static <T> Iterable<T> toIterable(@Nonnull PageProvider<T> pageProvider, int batchSize) {
        return new PagedIterable<>(requireNonNull(pageProvider, "pageProvider"), batchSize);
    }

    /**
     * A utility method to convert a {@link PageProvider} of {@code T} to a {@link Stream} of {@code T}
     *
     * @param pageProvider a page provider
     * @param batchSize    the size of the page to fetch
     * @return a {@link Stream} of objects of type {@code T}
     * @since 4.8
     */
    @Nonnull
    public static <T> Stream<T> toStream(@Nonnull PageProvider<T> pageProvider, int batchSize) {
        return MoreStreams.streamIterable(toIterable(pageProvider, batchSize));
    }
}
