package com.nextbreakpoint.shop.designs.persistence;

import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import com.google.common.util.concurrent.ListenableFuture;
import com.nextbreakpoint.shop.common.model.DesignDocument;
import com.nextbreakpoint.shop.designs.Store;
import com.nextbreakpoint.shop.designs.model.ListDesignsRequest;
import com.nextbreakpoint.shop.designs.model.ListDesignsResponse;
import com.nextbreakpoint.shop.designs.model.LoadDesignRequest;
import com.nextbreakpoint.shop.designs.model.LoadDesignResponse;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import rx.Single;

import java.time.Instant;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;

public class CassandraStore implements Store {
    private final Logger logger = LoggerFactory.getLogger(CassandraStore.class.getName());

    private static final String ERROR_LOAD_DESIGN = "An error occurred while loading a design";
    private static final String ERROR_LIST_DESIGNS = "An error occurred while loading designs";

    private static final String SELECT_DESIGN = "SELECT * FROM DESIGNS_VIEW WHERE DESIGN_UUID = ?";
    private static final String SELECT_DESIGNS = "SELECT DESIGN_UUID, DESIGN_CHECKSUM, DESIGN_TIMESTAMP FROM DESIGNS_VIEW";

    private static final int EXECUTE_TIMEOUT = 10;

    private final Supplier<Session> supplier;

    private Session session;

    private ListenableFuture<PreparedStatement> selectDesign;
    private ListenableFuture<PreparedStatement> selectDesigns;

    public CassandraStore(Supplier<Session> supplier) {
        this.supplier = Objects.requireNonNull(supplier);
    }

    @Override
    public Single<LoadDesignResponse> loadDesign(LoadDesignRequest request) {
        return withSession()
                .flatMap(session -> doLoadDesign(session, request))
                .doOnError(err -> handleError(ERROR_LOAD_DESIGN, err));
    }

    @Override
    public Single<ListDesignsResponse> listDesigns(ListDesignsRequest request) {
        return withSession()
                .flatMap(session -> doListDesigns(session, request))
                .doOnError(err -> handleError(ERROR_LIST_DESIGNS, err));
    }

    private Single<Session> withSession() {
        if (session == null) {
            session = supplier.get();
            if (session == null) {
                return Single.error(new RuntimeException("Cannot create session"));
            }
            selectDesign = session.prepareAsync(SELECT_DESIGN);
            selectDesigns = session.prepareAsync(SELECT_DESIGNS);
        }
        return Single.just(session);
    }

    private Single<ResultSet> getResultSet(ResultSetFuture rsf) {
        return Single.fromCallable(() -> rsf.getUninterruptibly(EXECUTE_TIMEOUT, TimeUnit.SECONDS));
    }

    private Single<LoadDesignResponse> doLoadDesign(Session session, LoadDesignRequest request) {
        return Single.from(selectDesign)
                .map(pst -> pst.bind(makeLoadParams(request)))
                .map(session::executeAsync)
                .flatMap(this::getResultSet)
                .map(rs -> toDesignDocuments(rs, this::getDesignDocument))
                .map(documents -> documents.stream().findFirst().orElse(null))
                .map(document -> new LoadDesignResponse(request.getUuid(), document));
    }

    private Single<ListDesignsResponse> doListDesigns(Session session, ListDesignsRequest request) {
        return Single.from(selectDesigns)
                .map(PreparedStatement::bind)
                .map(session::executeAsync)
                .flatMap(this::getResultSet)
                .map(rs -> toDesignDocuments(rs, this::getMinimalDesignDocument))
                .map(ListDesignsResponse::new);
    }

    private List<DesignDocument> toDesignDocuments(ResultSet rs, Function<Row, DesignDocument> mapper) {
        final List<DesignDocument> documents = new ArrayList<>();
        final Iterator<Row> iter = rs.iterator();
        if (iter.hasNext()) {
            if (rs.getAvailableWithoutFetching() >= 100 && !rs.isFullyFetched()) {
                rs.fetchMoreResults();
            }
            final Row row = iter.next();
            documents.add(mapper.apply(row));
        }
        return documents;
    }

    private DesignDocument getDesignDocument(Row row) {
        final String uuid = row.getUUID("DESIGN_UUID").toString();
        final String json = row.getString("DESIGN_JSON");
        final String checksum = row.getString("DESIGN_CHECKSUM");
        final Instant timestamp = row.getTimestamp("DESIGN_TIMESTAMP").toInstant();
        return new DesignDocument(uuid, json, checksum, formatDate(timestamp));
    }

    private DesignDocument getMinimalDesignDocument(Row row) {
        final String uuid = row.getUUID("DESIGN_UUID").toString();
        final String checksum = row.getString("DESIGN_CHECKSUM");
        final Instant timestamp = row.getTimestamp("DESIGN_TIMESTAMP").toInstant();
        return new DesignDocument(uuid, null, checksum, formatDate(timestamp));
    }

    private Object[] makeLoadParams(LoadDesignRequest request) {
        return new Object[] { request.getUuid() };
    }

    private void handleError(String message, Throwable err) {
        logger.error(message, err);
    }

    private String formatDate(Instant instant) {
        return DateTimeFormatter.ISO_INSTANT.format(instant);
    }
}

@@ -3486,32 +3486,35 @@
 ignDocuments(rs,
+ 1,
  this::getDesign
@@ -4023,16 +4023,19 @@
 ents(rs,
+ 0,
  this::g
@@ -4179,16 +4179,27 @@
 tSet rs,
+ int limit,
  Functio
@@ -4355,18 +4355,21 @@
         
-if
+while
  (iter.h
@@ -4599,24 +4599,119 @@
 pply(row));%0A
+            if (limit %3E 0 && documents.size() %3E= limit) %7B%0A                break;%0A            %7D%0A
         %7D%0A  

fix query