package io.delphiplatform.api.v3.bigtable;

import com.google.cloud.bigtable.data.v2.models.Query;
import com.google.cloud.bigtable.data.v2.models.Row;
import com.google.common.collect.ImmutableMap;

import org.apache.commons.lang3.tuple.Pair;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
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.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.SpotifyTrackStream;
import io.delphiplatform.api.v3.bigtable.entity.TrackStream;
import io.delphiplatform.api.v3.bigtable.processing.PaginationService;
import io.delphiplatform.api.v3.bigtable.processing.StreamDatesConverter;
import io.delphiplatform.api.v3.bigtable.processing.StreamDatesSortingService;
import io.delphiplatform.api.v3.bigtable.reader.BigtableColumns;
import io.delphiplatform.api.v3.constant.BigtableTableName;
import io.delphiplatform.api.v3.constant.DspConstants;
import io.delphiplatform.api.v3.model.StreamDate;
import io.delphiplatform.api.v3.view.util.Params;
import lombok.extern.slf4j.Slf4j;

import static io.delphiplatform.api.util.CollectionUtils.isEmpty;
import static io.delphiplatform.api.v3.constant.CountryCodeConstant.WORLDWIDE;

@Slf4j
@Service
public class BigtableStreamDatesService {

    private static final int BASE_LIMIT = 10;
    private static final int MAX_LIMIT = 500;
    private static final int LIMIT_MULTIPLIER = 2;
    private static final int MIN_APPLE_STREAMS_AMOUNT = 3;
    private static final int MIN_STREAMS_AMOUNT = 0;
    private final BigtableQueryExecutor queryExecutor;
    private final BigtableRowReader rowReader;
    private final StreamDatesSortingService streamsSortingService;
    private final PaginationService paginationService;
    private final StreamDatesConverter streamDatesConverter;

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

    public BigtableStreamDatesService(BigtableQueryExecutor queryExecutor,
        BigtableRowReader rowReader,
        StreamDatesSortingService streamsSortingService,
        PaginationService paginationService,
        StreamDatesConverter streamDatesConverter) {
        this.queryExecutor = queryExecutor;
        this.rowReader = rowReader;
        this.streamsSortingService = streamsSortingService;
        this.paginationService = paginationService;
        this.streamDatesConverter = streamDatesConverter;

        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();

    }

    /**
     * Get first streaming dates based on requested Params object
     *
     * @param params containing isrc, start date, end date, dsps and sorting/pagination info
     * @return future containing list of StreamDate for each dsp-isrc combination from BigTable
     */
    public CompletableFuture<List<StreamDate>> getStreamDates(Params params) {
        CompletableFuture<Stream<List<TrackStream<?, ?>>>> firstStreamDataForIsrc = getFirstStreamData(params);

        return firstStreamDataForIsrc
            .thenApply(streamDatesConverter::convert)
            .thenApply(
                streamDates -> streamsSortingService.sort(streamDates, params.getSortBy(), params.getSortOrder()))
            .thenApply(streamDates -> paginationService.getPage(streamDates, params))
            .thenApply(streamDates -> streamDates.collect(Collectors.toList()));
    }

    /**
     * Get stream of List<TrackStream> containing information from first streaming date for each isrc-dsp combination. List may contain multiple
     * objects if there were multiple streaming from different sub-dsps (only for Amazon: prime, ad-supported, unlimited)
     */
    private CompletableFuture<Stream<List<TrackStream<?, ?>>>> getFirstStreamData(Params params) {
        Collection<String> dsps =
            params.getDsp() == null || params.getDsp().isEmpty() ? dspToTable.keySet() : params.getDsp();
        List<String> isrcs = params.getIsrc();

        List<CompletableFuture<List<TrackStream<?, ?>>>> firstTrackStreamsForIsrc = dsps.stream()
            .flatMap(dsp -> isrcs.stream().map(isrc -> Pair.of(dsp, isrc)))
            .map(dspAndIsrc -> getFirstStreamDataForIsrc(params, dspAndIsrc.getLeft(), dspAndIsrc.getRight()))
            .collect(Collectors.toList());

        return CompletableFuture.allOf(firstTrackStreamsForIsrc.toArray(new CompletableFuture[0]))
            .thenApply(ignored -> firstTrackStreamsForIsrc.stream()
                .map(this::resolveFuture)
                .filter(streams -> !CollectionUtils.isEmpty(streams)));
    }

    /**
     * Get list of TrackStream from first streaming date for single dsp-isrc combination.
     */
    private CompletableFuture<List<TrackStream<?, ?>>> getFirstStreamDataForIsrc(Params params, String dsp,
        String isrc) {
        RequestRowKeyHandle range = generateRowKeySet(params, isrc);
        int initialLimit = getInitialLimit(dsp);
        CompletableFuture<List<TrackStream<?, ?>>> firstTrackStreams = queryTrackStreams(dsp, range, initialLimit, true);

        return firstTrackStreams.thenApply(firstStreams -> {
            if (isEmpty(firstStreams)) {
                return null;
            }
            return selectFirstStreamingData(firstStreams).orElseGet(() -> {
                int limit = initialLimit;
                List<TrackStream<?, ?>> currentTrackStreams = firstStreams;
                Optional<List<TrackStream<?, ?>>> firstStreamData = Optional.empty();

                while (firstStreamData.isEmpty() && currentTrackStreams.size() >= limit) {
                    limit = Math.min(limit * LIMIT_MULTIPLIER, MAX_LIMIT);
                    LocalDate currentStartDate = currentTrackStreams.stream()
                        .map(TrackStream::getDate)
                        .max(LocalDate::compareTo).orElse(params.getStartDate());
                    RequestRowKeyHandle currentRange = generateRowKeySet(params, isrc, currentStartDate);
                    currentTrackStreams = resolveFuture(queryTrackStreams(dsp, currentRange, limit, false));
                    firstStreamData = selectFirstStreamingData(currentTrackStreams);
                }
                return firstStreamData.orElse(Collections.emptyList());
            });
        });
    }

    private int getInitialLimit(String dsp) {
        return Objects.equals(dsp, DspConstants.AMAZON)
            ? BASE_LIMIT * DspConstants.AMAZON_SUB_DSPS.size()
            : BASE_LIMIT;
    }

    /**
     * Selects TrackStream's for the first date from provided list which may contain track streams for multiple days.
     * <p>
     * Select criteria: track steam contains streams info greater then 0, if there is a gap between first and second date in the list, then
     * consider first date as a test-stream date and select the second date.
     */
    private Optional<List<TrackStream<?, ?>>> selectFirstStreamingData(List<TrackStream<?, ?>> trackStreams) {
        if (CollectionUtils.isEmpty(trackStreams)) {
            return Optional.of(Collections.emptyList());
        }
        Map<LocalDate, List<TrackStream<?, ?>>> dateToStreams = trackStreams.stream()
            .filter(ts -> ts.getStreams(WORLDWIDE) != null && ts.getStreams(WORLDWIDE) > minStreamsAmount(ts.getDsp()))
            .collect(Collectors.groupingBy(TrackStream::getDate));

        if (dateToStreams.size() == 0) {
            return Optional.empty();
        }

        List<LocalDate> dates = dateToStreams.keySet().stream().sorted().collect(Collectors.toList());
        if (dates.size() == 1
            || DspConstants.APPLE.equals(dateToStreams.get(dates.get(0)).get(0).getDsp())
            || dates.get(0).equals(dates.get(1).minusDays(1))) {
            return Optional.of(dateToStreams.get(dates.get(0)));
        } else {
            return Optional.of(dateToStreams.get(dates.get(1)));
        }
    }

    private long minStreamsAmount(String dsp) {
        return DspConstants.APPLE.equals(dsp) ? MIN_APPLE_STREAMS_AMOUNT : MIN_STREAMS_AMOUNT;
    }

    private RequestRowKeyHandle generateRowKeySet(Params params, String isrc) {
        return generateRowKeySet(params, isrc, params.getStartDate());
    }

    private RequestRowKeyHandle generateRowKeySet(Params params, String isrc, LocalDate startDate) {
        try {
            return new RowKeyGenerator(
                params.toBuilder()
                    .isrc(Collections.singletonList(isrc))
                    .startDate(startDate)
                    .build(),
                RowKeyGroup.STREAM,
                RowKeyRequestType.RANGE
            ).getRequestRowKeyHandles().get(0);
        } catch (NotSupportedException e) {
            throw new RuntimeException(String.format("Cannot generate ranges for params=%s,isrc=%s,startDate=%s",
                params, isrc, startDate), e);
        }
    }

    /**
     * Query track stream from BigTable based on isrcRange and limit
     */
    private CompletableFuture<List<TrackStream<?, ?>>> queryTrackStreams(String dsp, RequestRowKeyHandle rowKey,
        int limit, boolean executeInAnotherThread) {
        Query query = Query.create(dspToTable.get(dsp));

        Pair<String, String> isrcRange = rowKey.getRowKeyRange();
        query.range(isrcRange.getLeft(), isrcRange.getRight());
        query.limit(limit);

        CompletableFuture<List<Row>> rowsForIsrc = executeInAnotherThread
            ? queryExecutor.executeQuery(query, Collections.singletonList(BigtableColumns.DEMOGRAPHICS))
            : queryExecutor.executeQuerySync(query, Collections.singletonList(BigtableColumns.DEMOGRAPHICS));

        return rowsForIsrc.thenApply(rowsStream -> rowsStream.stream()
            .<TrackStream<?, ?>>map(row -> rowReader.readRow(dspToEntityClass.get(dsp), row))
            .collect(Collectors.toList()));
    }

    private <T> T resolveFuture(CompletableFuture<T> s) {
        try {
            return s.get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException("Cannot get future result for BigTable query.", e);
        }
    }

}