package com.commercetools.pspadapter.payone.domain.ctp;

import com.commercetools.pspadapter.payone.domain.ctp.exceptions.NoCartLikeFoundException;
import io.sphere.sdk.carts.Cart;
import io.sphere.sdk.carts.queries.CartQuery;
import io.sphere.sdk.client.BlockingSphereClient;
import io.sphere.sdk.messages.GenericMessageImpl;
import io.sphere.sdk.messages.MessageDerivateHint;
import io.sphere.sdk.messages.expansion.MessageExpansionModel;
import io.sphere.sdk.messages.queries.MessageQuery;
import io.sphere.sdk.orders.Order;
import io.sphere.sdk.orders.queries.OrderQuery;
import io.sphere.sdk.payments.Payment;
import io.sphere.sdk.payments.messages.PaymentCreatedMessage;
import io.sphere.sdk.payments.messages.PaymentTransactionAddedMessage;
import io.sphere.sdk.payments.queries.PaymentByIdGet;
import io.sphere.sdk.queries.PagedQueryResult;
import io.sphere.sdk.queries.Query;

import java.time.ZonedDateTime;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

/**
 * @author fhaertig
 * @since 02.12.15
 */
//TODO: refactor class since it has mixed concerns (maybe MessageConsumer only)
public class CommercetoolsQueryExecutor {

    private BlockingSphereClient client;

    public CommercetoolsQueryExecutor(final BlockingSphereClient client) {
        this.client = client;
    }

    public PaymentWithCartLike getPaymentWithCartLike(final String paymentId) {
        final CompletionStage<Payment> payment = client.execute(PaymentByIdGet.of(paymentId));
        return getPaymentWithCartLike(paymentId, payment);
    }

    public PaymentWithCartLike getPaymentWithCartLike(String paymentId, CompletionStage<Payment> paymentFuture)  {
        final CompletionStage<PagedQueryResult<Order>> orderFuture = client.execute(OrderQuery.of().withPredicates(m -> m.paymentInfo().payments().id().is(paymentId)));
        final CompletionStage<PagedQueryResult<Cart>> cartFuture = client.execute(CartQuery.of().withPredicates(m -> m.paymentInfo().payments().id().is(paymentId)));

        final CompletionStage<PaymentWithCartLike> paymentWithCartLikeFuture = paymentFuture.thenCompose(payment ->
            orderFuture.thenCompose(orderResult -> {
                if (orderResult.getTotal() > 0) {
                    final Order order = orderResult.getResults().get(0);
                    return CompletableFuture.completedFuture(new PaymentWithCartLike(payment, order));
                } else {
                    return cartFuture.thenApply(cartResult -> {
                        if (cartResult.getTotal() > 0) {
                            return new PaymentWithCartLike(payment, cartResult.getResults().get(0));
                        } else {
                            throw new NoCartLikeFoundException();
                        }
                    });
                }
            }
        ));

        //TODO: refactor since BlockingClient is available
        try {
            return paymentWithCartLikeFuture
                .toCompletableFuture()
                .get(10, TimeUnit.SECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            final Throwable cause =
                e.getCause() != null && e instanceof ExecutionException
                    ? e.getCause()
                    : e;
            throw cause instanceof RuntimeException? (RuntimeException) cause : new CompletionException(cause);
        }
    }

    public void consumePaymentCreatedMessages(final ZonedDateTime sinceDate, final Consumer<Payment> paymentConsumer) {
        consumeAllMessages(sinceDate, paymentConsumer, PaymentCreatedMessage.MESSAGE_HINT);
    }

    public void consumePaymentTransactionAddedMessages(final ZonedDateTime sinceDate, final Consumer<Payment> paymentConsumer) {
        consumeAllMessages(sinceDate, paymentConsumer, PaymentTransactionAddedMessage.MESSAGE_HINT);
    }

    private <T extends GenericMessageImpl<Payment>> void consumeAllMessages(
            final ZonedDateTime sinceDate,
            final Consumer<Payment> paymentConsumer,
            final MessageDerivateHint<T> messageHint) {
        final MessageQuery baseQuery = MessageQuery.of()
            .withPredicates(m -> m.createdAt().isGreaterThanOrEqualTo(sinceDate))
            .withSort(m -> m.createdAt().sort().asc())
            .withExpansionPaths(MessageExpansionModel::resource)
            .withLimit(500); // Maximum

        long processed = 0, total = 0;

        do {
            Query<T> query = baseQuery
                .withOffset(processed)
                .forMessageType(messageHint);
            final PagedQueryResult<T> result = client.executeBlocking(query);

            result.getResults()
                    .stream()
                    .filter(msg -> msg.getResource().getObj() != null)
                    .forEach(msg -> paymentConsumer.accept(msg.getResource().getObj()));

            processed = result.getOffset() + result.size();
            total = result.getTotal();
        } while (processed < total);
    }

}

@@ -266,24 +266,71 @@
 hereClient;%0A
+import io.sphere.sdk.client.SphereClientUtils;%0A
 import io.sp
@@ -1047,194 +1047,51 @@
 tion
-Exception;%0Aimport java.util.concurrent.CompletionStage;%0Aimport java.util.concurrent.ExecutionException;%0Aimport java.util.concurrent.TimeUnit;%0Aimport java.util.concurrent.TimeoutException
+Stage;%0Aimport java.util.concurrent.TimeUnit
 ;%0Aim
@@ -1293,16 +1293,148 @@
 ecutor %7B
+%0A    // TODO jw: extract to use the same value as the ServiceFactory%0A    private static final long DEFAULT_CTP_CLIENT_TIMEOUT = 10L;
 %0A%0A    pr
@@ -3121,573 +3121,119 @@
     
-//TODO: refactor since BlockingClient is available%0A        try %7B%0A            return paymentWithCartLikeFuture%0A                .toCompletableFuture()%0A                .get(10, TimeUnit.SECONDS);%0A        %7D catch (InterruptedException %7C ExecutionException %7C TimeoutException e) %7B%0A            final Throwable cause =%0A                e.getCause() != null && e instanceof ExecutionException%0A                    ? e.getCause()%0A                    : e;%0A            throw cause instanceof RuntimeException? (RuntimeException) cause : new CompletionException(cause);%0A        %7D
+return SphereClientUtils.blockingWait(paymentWithCartLikeFuture, DEFAULT_CTP_CLIENT_TIMEOUT, TimeUnit.SECONDS);
 %0A   

Use SphereClientUtils' implementation for blocking wait