package io.delphiplatform.api.v3.bigtable;

import com.google.common.collect.ImmutableMap;

import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.transaction.NotSupportedException;

import io.delphiplatform.api.exception.InvalidParameterException;
import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.util.logging.LoggingUtils;
import io.delphiplatform.api.v3.bigtable.config.RowKeyGroup;
import io.delphiplatform.api.v3.bigtable.config.RowKeyRequestType;
import io.delphiplatform.api.v3.bigtable.config.RequestRowKeyHandle;
import io.delphiplatform.api.v3.bigtable.entity.AmazonTrackStream;
import io.delphiplatform.api.v3.bigtable.entity.AppleTrackStream;
import io.delphiplatform.api.v3.bigtable.entity.SpotifyTrackNewListenersStream;
import io.delphiplatform.api.v3.bigtable.entity.SpotifyTrackStream;
import io.delphiplatform.api.v3.bigtable.entity.TrackStream;
import io.delphiplatform.api.v3.bigtable.processing.PaginationService;
import io.delphiplatform.api.v3.bigtable.processing.StreamsAggregator;
import io.delphiplatform.api.v3.bigtable.processing.StreamsConverter;
import io.delphiplatform.api.v3.bigtable.processing.StreamsSortingService;
import io.delphiplatform.api.v3.bigtable.reader.BigtableColumns;
import io.delphiplatform.api.v3.constant.ApplicationConstants;
import io.delphiplatform.api.v3.constant.BigtableTableName;
import io.delphiplatform.api.v3.constant.DspConstants;
import io.delphiplatform.api.v3.constant.RequestConstants;
import io.delphiplatform.api.v3.model.IncludeStreams;
import io.delphiplatform.api.v3.model.StreamModel;
import io.delphiplatform.api.v3.view.util.Params;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class BigtableStreamsService {

    private final BigtableQueryExecutor queryExecutor;
    private final BigtableRowReader rowReader;
    private final StreamsConverter streamsConverter;
    private final StreamsAggregator streamsAggregator;
    private final StreamsSortingService streamsSortingService;
    private final PaginationService paginationService;

    private final Map<String, String> dspToTable;
    private final Map<String, Class<? extends TrackStream<?, ?>>> dspToEntityClass;
    private final Map<String, Class<? extends TrackStream<?, ?>>> dspToNewListenersEntityClass;

    public BigtableStreamsService(BigtableQueryExecutor queryExecutor, BigtableRowReader rowReader,
        StreamsConverter streamsConverter,
        StreamsAggregator streamsAggregator,
        StreamsSortingService streamsSortingService,
        PaginationService paginationService) {
        this.queryExecutor = queryExecutor;
        this.rowReader = rowReader;
        this.streamsConverter = streamsConverter;
        this.streamsAggregator = streamsAggregator;
        this.streamsSortingService = streamsSortingService;
        this.paginationService = paginationService;

        dspToTable = ImmutableMap.<String, String>builder()
            .put(DspConstants.AMAZON, BigtableTableName.AMAZON_STREAMS)
            .put(DspConstants.APPLE, BigtableTableName.APPLE_STREAMS)
            .put(DspConstants.SPOTIFY, BigtableTableName.SPOTIFY_STREAMS)
            .build();

        dspToEntityClass = ImmutableMap.<String, Class<? extends TrackStream<?, ?>>>builder()
            .put(DspConstants.AMAZON, AmazonTrackStream.class)
            .put(DspConstants.APPLE, AppleTrackStream.class)
            .put(DspConstants.SPOTIFY, SpotifyTrackStream.class)
            .build();

        dspToNewListenersEntityClass = ImmutableMap.<String, Class<? extends TrackStream<?, ?>>>builder()
            .put(DspConstants.SPOTIFY, SpotifyTrackNewListenersStream.class)
            .build();

    }

    public CompletableFuture<List<StreamModel>> getRows(Params params) throws NotSupportedException {
        if (CollectionUtils.isNotEmpty(params.getDsp())) {
            List<String> wrongDsps = params.getDsp().stream().filter(Predicate.not(dspToTable::containsKey))
                .collect(Collectors.toList());

            if (CollectionUtils.isNotEmpty(wrongDsps)) {
                throw new InvalidParameterException(
                    RequestConstants.DSP, Stream.of(getAvailableDsps(params.isNewListeners())),
                    StringUtils.join(wrongDsps, ApplicationConstants.COMMA));
            }
        }

        RowKeyGenerator rowKeyGenerator = new RowKeyGenerator(params, RowKeyGroup.STREAM, RowKeyRequestType.RANGE);
        List<RequestRowKeyHandle> rowKeys = rowKeyGenerator.getRequestRowKeyHandles();

        Collection<String> dspsToQuery =
            CollectionUtils.isEmpty(params.getDsp()) ? getAvailableDsps(params.isNewListeners())
                : params.getDsp();
        CompletableFuture<Stream<TrackStream<?, ?>>> allRows = getAllStreams(rowKeys, dspsToQuery,
            params.getIncludeStreams(), params.isNewListeners());

        return allRows
            .thenApply(streams -> streamsConverter.convertForCountryCode(params.getCountryCode(), streams, params.getIncludeStreams()))
            .thenApply(streams -> populateArtistId(streams, params))
            .thenApply(streams -> streamsAggregator.groupByAggregate(streams, params))
            .thenApply(streams -> streamsSortingService.sort(streams, params.getSortBy(), params.getSortOrder()))
            .thenApply(streams -> paginationService.getPage(streams, params))
            .thenApply(streams -> streams.collect(Collectors.toList()));
    }

    private Stream<StreamModel> populateArtistId(Stream<StreamModel> models, Params params) {
        String artistId = params.getArtistId();
        boolean replaceAnyArtistIdToTargetOne = params.getAdditionalArtistIds() != null;

        if (artistId != null) {
            return models.peek(model -> {
                boolean modelConvertedFromRowWithoutArtistId = model.getArtistId() == null;

                if (modelConvertedFromRowWithoutArtistId || replaceAnyArtistIdToTargetOne) {
                    model.artistId(artistId);
                }
            });
        } else {
            return models;
        }
    }

    private Set<String> getAvailableDsps(boolean newListeners) {
        if (newListeners) {
            return dspToNewListenersEntityClass.keySet();
        }
        return dspToTable.keySet();
    }

    private CompletableFuture<Stream<TrackStream<?, ?>>> getAllStreams(
        List<RequestRowKeyHandle> itemKeys,
        Collection<String> dsps,
        Set<IncludeStreams> includeStreams, boolean isNewListeners
    ) {
        StopWatch methodExecutionStopwatch = new StopWatch();
        methodExecutionStopwatch.start();

        List<String> ignoredFields = getIgnoredFields(includeStreams, isNewListeners);
        List<CompletableFuture<Stream<TrackStream<?, ?>>>> rows = dsps.stream()
            .filter(dsp -> !isNewListeners || dspToNewListenersEntityClass.containsKey(dsp))
            .map(dsp -> queryExecutor.executePartitioned(dspToTable.get(dsp), itemKeys, ignoredFields)
                .<Stream<TrackStream<?, ?>>>thenApply(rowStream -> rowStream
                    .map(row -> rowReader.readRow(getEntityClass(dsp, isNewListeners), row))))
            .collect(Collectors.toList());

        return CompletableFuture.allOf(rows.toArray(new CompletableFuture[0]))
            .thenApply(ignored -> {
                LoggingUtils.stopAndLogManualPerformanceMeasurement(methodExecutionStopwatch,
                    "BigtableStreamsService.getAllStreams: BT query");

                return rows.stream().flatMap(s -> {
                    try {
                        return s.get();
                    } catch (InterruptedException | ExecutionException e) {
                        throw new RuntimeException("Cannot get future result for BigTable query.", e);
                    }
                });
            });
    }

    private List<String> getIgnoredFields(Set<IncludeStreams> includeStreams, boolean isNewListeners) {
        List<String> ignoredFields = new ArrayList<>();
        if (isNewListeners) {
            ignoredFields.add(BigtableColumns.STREAMS);
            ignoredFields.add(BigtableColumns.DEMOGRAPHICS);
            if (!IncludeStreams.isDemographicsIncluded(includeStreams)) {
                ignoredFields.add(BigtableColumns.NEW_LISTENERS_DEMOGRAPHICS);
            }
        } else {
            ignoredFields.add(BigtableColumns.NEW_LISTENERS_STREAMS);
            ignoredFields.add(BigtableColumns.NEW_LISTENERS_DEMOGRAPHICS);
            if (!IncludeStreams.isDemographicsIncluded(includeStreams)) {
                ignoredFields.add(BigtableColumns.DEMOGRAPHICS);
            }
        }
        return ignoredFields;
    }

    private Class<? extends TrackStream<?, ?>> getEntityClass(String dsp, boolean isNewListeners) {
        if (isNewListeners) {
            return dspToNewListenersEntityClass.get(dsp);
        }
        return dspToEntityClass.get(dsp);
    }

}