package io.delphiplatform.api.v3.bigtable;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Streams;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.transaction.NotSupportedException;

import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.util.model.ModelUtils;
import io.delphiplatform.api.util.logging.LoggingUtils;
import io.delphiplatform.api.v3.bigtable.config.RowKeyAggPeriod;
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.PlaylistPublicDaily;
import io.delphiplatform.api.v3.bigtable.entity.PlaylistPublicPosition;
import io.delphiplatform.api.v3.bigtable.reader.BigtableEntityReader.EntityReadingParams;
import io.delphiplatform.api.v3.constant.BigtableTableName;
import io.delphiplatform.api.v3.constant.DspConstants;
import io.delphiplatform.api.v3.view.util.Params;

@Service
public class BigtableTrackPositionPublicQueryExecutor {

    private final BigtableQueryExecutor queryExecutor;
    private final BigtableRowReader rowReader;

    private final Map<String, String> factPublicPlaylistDailyTableMap;

    @Autowired
    public BigtableTrackPositionPublicQueryExecutor(BigtableQueryExecutor queryExecutor, BigtableRowReader rowReader) {
        this.queryExecutor = queryExecutor;
        this.rowReader = rowReader;

        factPublicPlaylistDailyTableMap = ImmutableMap.<String, String>builder()
            .put(DspConstants.APPLE, BigtableTableName.PLAYLIST_APPLE_DAILY)
            .put(DspConstants.SPOTIFY, BigtableTableName.PLAYLIST_SPOTIFY_DAILY)
            .build();
    }

    public CompletableFuture<Stream<PlaylistPublicPosition>> getAllPublicPlaylistTrackPositions(Params params)
        throws NotSupportedException {
        return getAllPublicPlaylistTrackPositions(params, null, Collections.emptyList(), Collections.emptyList());
    }

    public CompletableFuture<Stream<PlaylistPublicPosition>> getAllPublicPlaylistTrackPositions(
        Params params,
        EnumSet<EntityReadingParams> entityReadingParams
    ) throws NotSupportedException {
        return getAllPublicPlaylistTrackPositions(params, entityReadingParams, Collections.emptyList(), Collections.emptyList());
    }

    public CompletableFuture<Stream<PlaylistPublicPosition>> getAllPublicPlaylistTrackPositions(
        Params params,
        EnumSet<EntityReadingParams> entityReadingParams,
        Collection<String> spotifyAppleIgnoredColumns,
        Collection<String> amazonIgnoredColumns
    ) throws NotSupportedException {
        StopWatch methodExecutionStopwatch = new StopWatch();
        methodExecutionStopwatch.start();

        boolean isNotFilteredByDspOrPlaylist = CollectionUtils.isEmpty(params.getDsp()) && CollectionUtils.isEmpty(params.getPlaylistId());
        boolean isFilteredByDsp = CollectionUtils.isNotEmpty(params.getDsp());
        boolean isFilteredByPlaylist = CollectionUtils.isNotEmpty(params.getPlaylistId());

        boolean needToQuerySpotify = isNotFilteredByDspOrPlaylist
            || (isFilteredByDsp && params.getDsp().contains(DspConstants.SPOTIFY))
            || (isFilteredByPlaylist && params.getPlaylistId().stream().anyMatch(
            pId -> ModelUtils.isPlaylistOfDsp(pId, DspConstants.SPOTIFY)));

        boolean needToQueryApple = isNotFilteredByDspOrPlaylist
            || (isFilteredByDsp && params.getDsp().contains(DspConstants.APPLE))
            || (isFilteredByPlaylist && params.getPlaylistId().stream().anyMatch(
            pId -> ModelUtils.isPlaylistOfDsp(pId, DspConstants.APPLE)));

        boolean needToQueryAmazon = isNotFilteredByDspOrPlaylist
            || (isFilteredByDsp && params.getDsp().contains(DspConstants.AMAZON))
            || (isFilteredByPlaylist && params.getPlaylistId().stream().anyMatch(
            pId -> ModelUtils.isPlaylistOfDsp(pId, DspConstants.AMAZON)));

        CompletableFuture<Stream<PlaylistPublicPosition>> spotifyAndApplePositions;
        CompletableFuture<Stream<PlaylistPublicPosition>> amazonPositions;

        if (needToQuerySpotify || needToQueryApple) {
            List<String> dsps = new ArrayList<>();
            if (needToQuerySpotify) {
                dsps.add(DspConstants.SPOTIFY);
            }
            if (needToQueryApple) {
                dsps.add(DspConstants.APPLE);
            }

            List<String> spotifyAndApplePlaylistIdsToQuery = isFilteredByPlaylist
                ? params.getPlaylistId().stream()
                .filter(
                    pId -> ModelUtils.isPlaylistOfDsp(pId, DspConstants.SPOTIFY)
                        || ModelUtils.isPlaylistOfDsp(pId, DspConstants.APPLE)
                ).collect(Collectors.toList())
                : null;

            //!spotifyAndApplePlaylistIdsToQuery.isEmpty() check is to skip BT request for case when filters are like
            // e.g. dsp=spotify,amazon&playlistId=amazon_123 (so both dsp/playlistId filters are present but no spotify playlists in filter)
            if (!isFilteredByPlaylist || !spotifyAndApplePlaylistIdsToQuery.isEmpty()) {
                Params spotifyAndAppleParams = params.toBuilder()
                    .dsp(dsps)
                    .playlistId(spotifyAndApplePlaylistIdsToQuery)
                    .build();

                RowKeyGenerator spotifyAndAppleRowKeyGenerator =
                    new RowKeyGenerator(spotifyAndAppleParams, RowKeyGroup.PLAYLIST, RowKeyAggPeriod.DAY, RowKeyRequestType.RANGE);
                List<RequestRowKeyHandle> spotifyAndAppleRowKeys = spotifyAndAppleRowKeyGenerator.getRequestRowKeyHandles();

                //for Spotify/Apple - query 'playlist_tracks' BT table
                spotifyAndApplePositions = queryExecutor.executePartitioned(BigtableTableName.PLAYLIST_TRACKS, spotifyAndAppleRowKeys,
                        spotifyAppleIgnoredColumns)
                    .thenApply(rowStream -> rowReader.readRows(PlaylistPublicPosition.class, rowStream, entityReadingParams));
            } else {
                spotifyAndApplePositions = CompletableFuture.completedFuture(Stream.empty());
            }
        } else {
            spotifyAndApplePositions = CompletableFuture.completedFuture(Stream.empty());
        }

        if (needToQueryAmazon) {

            List<String> amazonPlaylistIdsToQuery = isFilteredByPlaylist
                ? params.getPlaylistId().stream()
                .filter(
                    pId -> ModelUtils.isPlaylistOfDsp(pId, DspConstants.AMAZON)
                ).collect(Collectors.toList())
                : null;

            //!amazonPlaylistIdsToQuery.isEmpty() check is to skip BT request for case when filters are like
            // e.g. dsp=spotify,amazon&playlistId=spotify_123 (so both dsp/playlistId filters are present but no amazon playlists in filter)
            if (!isFilteredByPlaylist || !amazonPlaylistIdsToQuery.isEmpty()) {
                Params amazonParams = params.toBuilder()
                    .dsp(Collections.singletonList(DspConstants.AMAZON))
                    .playlistId(amazonPlaylistIdsToQuery)
                    .build();

                RowKeyGenerator amazonRowKeyGenerator = new RowKeyGenerator(amazonParams, RowKeyGroup.PLAYLIST,
                    RowKeyAggPeriod.DAY, RowKeyRequestType.RANGE);
                List<RequestRowKeyHandle> amazonRowKeys = amazonRowKeyGenerator.getRequestRowKeyHandles();

                //for Amazon - query common 'playlists' BT table but with single dsp filter
                amazonPositions = queryExecutor.executePartitioned(BigtableTableName.PLAYLISTS, amazonRowKeys, amazonIgnoredColumns)
                    .thenApply(rowStream -> rowReader.readRows(PlaylistPublicPosition.class, rowStream, entityReadingParams));
            } else {
                amazonPositions = CompletableFuture.completedFuture(Stream.empty());
            }
        } else {
            amazonPositions = CompletableFuture.completedFuture(Stream.empty());
        }

        return CompletableFuture.allOf(
            spotifyAndApplePositions,
            amazonPositions
        ).thenApply(ignored -> {
            try {
                LoggingUtils.stopAndLogManualPerformanceMeasurement(methodExecutionStopwatch,
                    "BigtableTrackPositionPublicQueryExecutor.getAllPublicPlaylistTrackPositions: BT query");

                return Streams.concat(spotifyAndApplePositions.get(), amazonPositions.get());
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException("Cannot get future result for BigTable query.", e);
            }
        });
    }

    public CompletableFuture<Stream<PlaylistPublicDaily>> getPlaylistDailyStats(Params params)
        throws NotSupportedException {
        RowKeyGenerator rowKeyGenerator = new RowKeyGenerator(params, RowKeyGroup.FACT_PLAYLIST_DAILY,
            RowKeyAggPeriod.DAY, RowKeyRequestType.RANGE);
        List<RequestRowKeyHandle> rowKeys = rowKeyGenerator.getRequestRowKeyHandles();

        if (!factPublicPlaylistDailyTableMap.containsKey(params.getDspId())) {
            throw new RuntimeException(params.getDspId() + " is not supported for querying playlist daily facts.");
        }

        return queryExecutor.executePartitioned(factPublicPlaylistDailyTableMap.get(params.getDspId()), rowKeys)
            .thenApply(rowStream -> rowReader.readRows(PlaylistPublicDaily.class, rowStream));
    }
}