package io.delphiplatform.api.util;

import com.google.common.collect.Lists;

import org.springframework.data.jpa.domain.Specification;

import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.persistence.criteria.CriteriaBuilder;
import javax.persistence.criteria.CriteriaBuilder.Coalesce;
import javax.persistence.criteria.Expression;
import javax.persistence.criteria.From;
import javax.persistence.criteria.Join;
import javax.persistence.criteria.JoinType;
import javax.persistence.criteria.Path;
import javax.persistence.criteria.Root;

import io.delphiplatform.api.v3.rdb.service.dto.JpaPropertyPath;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class JpaUtils {

    public static <T> Path<T> getPath(Path<T> path, List<String> propertyPath) {
        return getWithProperties(path, propertyPath);
    }

    public static <T> Path<T> getPathWithJoins(From<?, T> from, JpaPropertyPath propertyPath) {
        List<String> joins = propertyPath.getJoins();
        List<String> propertyNames = propertyPath.getProperties();

        From<?, T> resultFrom = from;
        for (String joinName : joins) {
            resultFrom = getJoin(resultFrom, joinName);
        }

        return getWithProperties(resultFrom, propertyNames);
    }

    private static <T> Path<T> getWithProperties(Path<T> root, List<String> propertyNames) {
        Path<T> resultExpression = root;
        for (String propertyName : propertyNames) {
            resultExpression = resultExpression.get(propertyName);
        }
        return resultExpression;
    }

    private static <T, E> Join<T, E> getJoin(From<?, T> resultFrom, String joinName) {
        Set<Join<T, ?>> existingJoins = resultFrom.getJoins();
        for (Join<T, ?> existingJoin : existingJoins) {
            if (existingJoin.getAttribute().getName().equals(joinName)) {
                return (Join<T, E>) existingJoin;
            }
        }
        return resultFrom.join(joinName, JoinType.LEFT);
    }

    public static <T> Specification<T> and(List<Specification<T>> specifications) {
        //noinspection ConstantConditions Specification::and is nullable only if both specs are null
        return specifications.stream()
            .filter(Objects::nonNull)
            .reduce(Specification::and)
            .orElse(null);
    }

    public static <T> Specification<T> or(Specification<T>... specifications) {
        return or(List.of(specifications));
    }

    public static <T> Specification<T> or(List<Specification<T>> specifications) {
        //noinspection ConstantConditions Specification::or is nullable only if both specs are null
        return specifications.stream()
            .filter(Objects::nonNull)
            .reduce(Specification::or)
            .orElse(null);
    }

    public static Expression<?> coalescePaths(CriteriaBuilder cb, Root<?> root, JpaPropertyPath firstPath,
        JpaPropertyPath secondPath) {
        Path<?> campaignProject = getPathWithJoins(root, firstPath);
        Path<?> linkProject = getPathWithJoins(root, secondPath);
        return cb.coalesce(linkProject, campaignProject);
    }

    public static Expression<?> coalescePaths(CriteriaBuilder cb, Root<?> root, JpaPropertyPath... jpaPaths) {
        Coalesce<Object> result = cb.coalesce();

        List<? extends Path<?>> paths = Arrays.stream(jpaPaths)
            .map(p -> getPathWithJoins(root, p))
            .collect(Collectors.toList());
        for (Path<?> path : paths) {
            result = result.value(path);
        }

        return result;
    }

    public static <T, A, R, I> R splitRepositoryMethodCall(Collection<I> ids,
        Function<Collection<I>, Collection<T>> repositoryMethod, Collector<T, A, R> collector) {

        int idColumnsCount = 1;
        if (!ids.isEmpty()) {
            I idColumn = ids.stream().findFirst().get();
            boolean isColumnSimpleType = idColumn instanceof String ||
                idColumn instanceof Long ||
                idColumn instanceof Integer;

            if (!isColumnSimpleType) {
                idColumnsCount = (int) Arrays.stream(idColumn.getClass().getDeclaredFields())
                    .filter(field -> !Modifier.isStatic(field.getModifiers()))
                    .count();
            }
        }

        int chunkSize = Short.MAX_VALUE / idColumnsCount;
        return Lists.partition(new ArrayList<>(ids), chunkSize)
            .parallelStream()
            .map(c -> repositoryMethod.apply(c))
            .flatMap(Collection::stream)
            .collect(collector);
    }

    public static <T, A, R, I> R splitExecutionByCollectionSize(Collection<I> ids,
        Function<Collection<I>, List<T>> fetchMethod,
        int size,
        Collector<T, A, R> collector) {

        List<CompletableFuture<Stream<T>>> partitions = Lists
            .partition(new ArrayList<>(ids), size)
            .stream()
            .map(c -> new CompletableFuture().supplyAsync(() -> fetchMethod.apply(c).stream()))
            .collect(Collectors.toList());

        return CompletableFuture.allOf(partitions.toArray(CompletableFuture[]::new))
            .thenApply(ignored -> partitions.stream().flatMap(future -> future.join()))
            .join()
            .collect(collector);
    }

    public static <T, F, A, R, I> R splitExecutionByCollectionSizeFiltered(Collection<I> ids,
        F filter,
        BiFunction<Collection<I>, F, Collection<T>> fetchMethod,
        int size,
        Collector<T, A, R> collector) {

        Collection<CompletableFuture<Stream<T>>> partitions = Lists
            .partition(new ArrayList<>(ids), size)
            .stream()
            .map(c -> new CompletableFuture().supplyAsync(() -> fetchMethod.apply(c, filter).stream()))
            .collect(Collectors.toList());

        return CompletableFuture.allOf(partitions.toArray(CompletableFuture[]::new))
            .thenApply(ignored -> partitions.stream().flatMap(future -> future.join()))
            .join()
            .collect(collector);
    }
}