/*
 * Copyright (C) 2012-2016 DuyHai DOAN
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package info.archinnov.achilles.internals.statements;

import static info.archinnov.achilles.internals.cql.TypeMapper.toJavaType;
import static info.archinnov.achilles.internals.utils.LoggerHelper.replaceByteBuffersByHexString;
import static java.lang.String.format;

import java.util.*;
import java.util.stream.IntStream;

import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.datastax.driver.core.*;
import com.datastax.driver.core.exceptions.TraceRetrievalException;

import info.archinnov.achilles.internals.options.Options;
import info.archinnov.achilles.internals.types.ResultSetWrapper;
import info.archinnov.achilles.logger.AchillesLoggers;

public interface StatementWrapper {
    Logger LOGGER = LoggerFactory.getLogger(StatementWrapper.class);

    EventComparator EVENT_TRACE_COMPARATOR = new EventComparator();
    Logger DML_LOGGER = LoggerFactory.getLogger(AchillesLoggers.ACHILLES_DML_STATEMENT);
    int RESULTS_LOG_DISPLAY_LIMIT = 10;

    Object[] getBoundValues();

    BoundStatement getBoundStatement();

    void applyOptions(Options options);

    void logDML();

    ResultSet logReturnResults(ResultSet resultSet);

    Row logReturnedRow(Row row);

    ResultSet logTrace(ResultSet resultSet);

    default void writeDMLStatementLog(Logger actualLogger, UUID queryId, String queryString, ConsistencyLevel consistencyLevel, Object[] boundValues, Object[] encodedValues) {
        if (actualLogger.isDebugEnabled()) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(format("Writing DML log for query %s with id %s", queryString, queryId));
            }
            actualLogger.debug("Query ID {} : [{}] with CONSISTENCY LEVEL [{}]",
                    queryId.toString(), queryString, consistencyLevel);
            if (ArrayUtils.isNotEmpty(boundValues)) {
                actualLogger.debug("\t Java bound values : {}", replaceByteBuffersByHexString(boundValues));
                actualLogger.debug("\t Encoded bound values : {}", replaceByteBuffersByHexString(encodedValues));
            }
        }
    }

    default void logReturnedResultsInternal(Logger actualLogger, UUID queryId, ResultSetWrapper resultSet) {
        final int availableWithoutFetching = resultSet.getAvailableWithoutFetching();
        StringBuilder results = new StringBuilder(format("Query ID %s results : \n", queryId));
        actualLogger.debug(resultSet.toString());
        for (int i = 0; i < Integer.min(availableWithoutFetching, RESULTS_LOG_DISPLAY_LIMIT); i++) {
            final Row row = resultSet.peek();
            appendRowDataToBuilder(row, row.getColumnDefinitions().asList(), results);
        }
        actualLogger.debug(results.toString());
    }

    default void logReturnedRowInternal(Logger actualLogger, UUID queryId, Row row) {
        StringBuilder results = new StringBuilder(format("Query ID %s row : \n", queryId));
        appendRowDataToBuilder(row, row.getColumnDefinitions().asList(), results);
        actualLogger.debug(results.toString());
    }

    default void appendRowDataToBuilder(Row row, List<ColumnDefinitions.Definition> columnsDef, StringBuilder builder) {
        StringJoiner joiner = new StringJoiner(", ", "\t", "\n");
        IntStream.range(0, columnsDef.size())
                .forEach(index -> {
                    final ColumnDefinitions.Definition def = columnsDef.get(index);
                    final Object value = extractValueFromRow(row, index, def.getType());
                    joiner.add(format("%s: %s", def.getName(), value));
                });
        builder.append(joiner.toString());
    }

    default Object extractValueFromRow(Row row, int index, DataType dataType) {
        final DataType.Name typeName = dataType.getName();
        switch (typeName) {
            case LIST:
            case SET:
            case MAP:
            case TUPLE:
            case CUSTOM:
                return row.getObject(index);
            default:
                return row.get(index, toJavaType(typeName));
        }
    }

    default void tracingInternal(Logger actualLogger, UUID queryId, ResultSet resultSet) {
        StringBuilder trace = new StringBuilder();
        if (actualLogger.isTraceEnabled()) {
            for (ExecutionInfo executionInfo : resultSet.getAllExecutionInfo()) {

                trace.append(format("\n\nTracing for Query ID {%s} at host {%s} with achieved consistency level {} \n", queryId.toString(), executionInfo.getQueriedHost(), executionInfo.getAchievedConsistencyLevel()));
                trace.append("****************************\n");
                trace.append(format("%1$-80s | %2$-16s | %3$-24s | %4$-20s\n", "Description", "Source", "Source elapsed in micros", "Thread name"));
                try {
                    final QueryTrace queryTrace = executionInfo.getQueryTrace();
                    if (queryTrace != null) {
                        final List<QueryTrace.Event> events = new ArrayList<>(queryTrace.getEvents());
                        Collections.sort(events, EVENT_TRACE_COMPARATOR);
                        for (QueryTrace.Event event : events) {
                            trace.append(format("%1$-80s | %2$-16s | %3$-24s | %4$-20s\n", event.getDescription(), event.getSource(), event.getSourceElapsedMicros(), event.getThreadName()));
                        }
                    }
                } catch (TraceRetrievalException e) {
                    final String queryString = getBoundStatement().preparedStatement().getQueryString();
                    trace.append(format(" ERROR: cannot retrieve trace for query {} because it may not be yet available", queryString));
                }
                trace.append("****************************\n\n");
            }

            actualLogger.trace(trace.toString());
        }
    }

    class EventComparator implements Comparator<QueryTrace.Event> {
        @Override
        public int compare(QueryTrace.Event event1, QueryTrace.Event event2) {
            return event1.getSource().toString().compareTo(event2.getSource().toString());
        }
    }
}

@@ -5016,20 +5016,18 @@
 uery ID 
-%7B
 %25s
-%7D
  at host
@@ -5031,12 +5031,10 @@
 ost 
-%7B
 %25s
-%7D
  wit
@@ -5062,18 +5062,18 @@
 y level 
-%7B%7D
+%25s
  %5Cn%22, qu

Fix typo in tracing logs