package io.delphiplatform.api.v3.bigtable;

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

import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
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.YouTubeVideoStats;
import io.delphiplatform.api.v3.bigtable.processing.PaginationService;
import io.delphiplatform.api.v3.bigtable.processing.YouTubeSummariesAggregator;
import io.delphiplatform.api.v3.bigtable.processing.YouTubeSummariesConverter;
import io.delphiplatform.api.v3.bigtable.processing.YouTubeSummariesSortingService;
import io.delphiplatform.api.v3.bigtable.propertyfilter.ResponsePropertyFilter;
import io.delphiplatform.api.v3.constant.BigtableTableName;
import io.delphiplatform.api.v3.model.video.GroupByVideo;
import io.delphiplatform.api.v3.model.video.Video;
import io.delphiplatform.api.v3.model.video.YouTubeMetricsCategory;
import io.delphiplatform.api.v3.model.video.YouTubeSummaryItem;
import io.delphiplatform.api.v3.rdb.service.VideoService;
import io.delphiplatform.api.v3.view.util.Params;
import lombok.extern.slf4j.Slf4j;

import static io.delphiplatform.api.util.CollectionUtils.contains;
import static io.delphiplatform.api.util.CollectionUtils.isEmpty;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.DEMOGRAPHICS;

@Slf4j
@Service
public class BigtableYouTubeVideoSummariesService {

    private final BigtableQueryExecutor queryExecutor;
    private final BigtableRowReader rowReader;
    private final YouTubeSummariesConverter youTubeSummariesConverter;
    private final VideoService videoService;
    private final YouTubeSummariesAggregator summariesAggregator;
    private final YouTubeSummariesSortingService sortingService;
    private final PaginationService paginationService;
    private final ResponsePropertyFilter responsePropertyFilter;

    public BigtableYouTubeVideoSummariesService(BigtableQueryExecutor queryExecutor, BigtableRowReader rowReader,
        YouTubeSummariesConverter youTubeSummariesConverter,
        VideoService videoService,
        YouTubeSummariesAggregator summariesAggregator,
        YouTubeSummariesSortingService sortingService,
        PaginationService paginationService,
        ResponsePropertyFilter responsePropertyFilter) {
        this.queryExecutor = queryExecutor;
        this.rowReader = rowReader;
        this.youTubeSummariesConverter = youTubeSummariesConverter;
        this.videoService = videoService;
        this.summariesAggregator = summariesAggregator;
        this.sortingService = sortingService;
        this.paginationService = paginationService;
        this.responsePropertyFilter = responsePropertyFilter;
    }

    /**
     * @param params request parameters
     * @return Row data from Bigtable
     * @throws NotSupportedException unsupported row key format
     */
    public CompletableFuture<List<YouTubeSummaryItem>> getRows(Params params) throws NotSupportedException {
        boolean groupByIsrcNotVideo = contains(params.getGroupByFieldsVideo(), GroupByVideo.ISRC) &&
            !contains(params.getGroupByFieldsVideo(), GroupByVideo.VIDEO_ID);
        if (groupByIsrcNotVideo) {
            return getRowsGroupedByIsrcNotVideo(params);
        }

        List<Video> videos;
        if (CollectionUtils.isNotEmpty(params.getProjectNumbers()) || CollectionUtils.isNotEmpty(
            params.getProductFamilyIds())) {
            videos = Collections.emptyList();
        } else {
            videos = getVideos(params);
            if (isEmpty(videos)) {
                return CompletableFuture.completedFuture(Collections.emptyList());
            }
        }

        ImmutableMap<String, Video> videosById = ImmutableMap.copyOf(videos.stream()
            .collect(Collectors.toMap(Video::getVideoId, Function.identity())));

        Params videoIdsParams = getVideoIdParams(videos, params);
        RowKeyGenerator rowKeyGenerator = new RowKeyGenerator(videoIdsParams, RowKeyGroup.VIDEO, RowKeyRequestType.RANGE);
        List<RequestRowKeyHandle> rowKeys = rowKeyGenerator.getRequestRowKeyHandles();

        CompletableFuture<Stream<YouTubeVideoStats>> allRows = getAllStats(rowKeys,
            YouTubeMetricsCategory.isDemographicsIncluded(params.getMetricCategories()));

        return allRows
            .thenApply(statsStream -> youTubeSummariesConverter.convert(statsStream, params.getCountryCode()))
            .thenApply(summaries -> summaries.peek(s -> {
                Video video = videosById.get(s.getDimensions().getVideoId());
                if (video != null) {
                    s.getDimensions().channelId(video.getChannelId());
                    s.getDimensions().contentType(video.getContentType());
                    s.getDimensions().dspVideoId(video.getDspVideoId());
                }

            }))
            .thenApply(summaries -> joinWithIsrcs(videosById, summaries, params))
            .thenApply(summaries -> summariesAggregator.groupByAggregate(summaries, params))
            .thenApply(summaries -> sortingService.sort(summaries, params.getSortBy(), params.getSortOrder()))
            .thenApply(summaries -> paginationService.getPage(summaries, params))
            .thenApply(summaries -> responsePropertyFilter.filterProperties(summaries, params.getMetricCategories()))
            .thenApply(summaries -> summaries.collect(Collectors.toList()));
    }

    /**
     * For the case when user provided group_by=isrc without group_by=video_id, we need to only get the video ISRCs and then query Bigtable.
     *
     * @param params request parameters
     * @return Row data from Bigtable
     * @throws NotSupportedException unsupported row key format
     */
    private CompletableFuture<List<YouTubeSummaryItem>> getRowsGroupedByIsrcNotVideo(Params params)
        throws NotSupportedException {
        Set<String> isrcs = videoService.getIsrcsForVideos(getVideoSearchParams(params));
        if (isEmpty(isrcs)) {
            return CompletableFuture.completedFuture(Collections.emptyList());
        }

        Params videoIsrcParams = getVideoIsrcParams(List.copyOf(isrcs), params);
        RowKeyGenerator rowKeyGenerator = new RowKeyGenerator(videoIsrcParams, RowKeyGroup.VIDEO, RowKeyRequestType.RANGE);
        List<RequestRowKeyHandle> rowKeys = rowKeyGenerator.getRequestRowKeyHandles();

        CompletableFuture<Stream<YouTubeVideoStats>> allRows = getAllStats(rowKeys,
            YouTubeMetricsCategory.isDemographicsIncluded(params.getMetricCategories()));

        return allRows
            .thenApply(statsStream -> youTubeSummariesConverter.convert(statsStream, params.getCountryCode()))
            .thenApply(summaries -> summariesAggregator.groupByAggregate(summaries, params))
            .thenApply(summaries -> sortingService.sort(summaries, params.getSortBy(), params.getSortOrder()))
            .thenApply(summaries -> paginationService.getPage(summaries, params))
            .thenApply(summaries -> responsePropertyFilter.filterProperties(summaries, params.getMetricCategories()))
            .thenApply(summaries -> summaries.collect(Collectors.toList()));
    }

    private Stream<YouTubeSummaryItem> joinWithIsrcs(Map<String, Video> videosById,
        Stream<YouTubeSummaryItem> summaries, Params params) {
        if (contains(params.getGroupByFieldsVideo(), GroupByVideo.ISRC)) {
            Map<String, ImmutableSet<String>> videoIdsToIsrcs = videoService.getVideoIdToIsrcs(videosById.keySet());

            return summariesAggregator.joinWithIsrc(summaries, videoIdsToIsrcs);

        } else {
            return summaries;
        }
    }

    /**
     * Get params used for querying VideoEntity. Removes date range, sorting, and pagination.
     */
    private Params getVideoSearchParams(Params params) {
        return Params.builder()
            .isrc(params.getIsrc())
            .artistId(params.getArtistId())
            .expandTo(params.getExpandTo())
            .youTubeContentTypes(params.getYouTubeContentTypes())
            .minPremiumUgcViews(params.getMinPremiumUgcViews())
            .videoIds(params.getVideoIds())
            .dsp(params.getDsp())
            .build();
    }

    /**
     * Modified params used for building row keys. Get a new Params object with injected ISRCs. Removes video_id, sorting, and pagination.
     */
    private Params getVideoIsrcParams(List<String> isrcs, Params params) {
        return Params.builder()
            .isrc(isrcs)
            .expandTo(params.getExpandTo())
            .youTubeContentTypes(params.getYouTubeContentTypes())
            .minPremiumUgcViews(params.getMinPremiumUgcViews())
            .startDate(params.getStartDate())
            .endDate(params.getEndDate())
            .dsp(params.getDsp())
            .build();
    }

    /**
     * Modified params used for building row keys. Get a new Params object based off of original with injected Videos.
     */
    private Params getVideoIdParams(List<Video> videos, Params params) {
        Set<String> formattedVideoIds = videos.stream()
            .map(Video::getVideoId)
            .collect(Collectors.toSet());

        return params.toBuilder().videoIds(formattedVideoIds).build();
    }

    private List<Video> getVideos(Params params) {
        return videoService.getAllVideosForFurtherProcessing(params);
    }

    private CompletableFuture<Stream<YouTubeVideoStats>> getAllStats(List<RequestRowKeyHandle> itemKeys,
        boolean isDemographicsIncluded) {
        CompletableFuture<Stream<Row>> rows = isDemographicsIncluded
            ? queryExecutor.executePartitioned(BigtableTableName.YOUTUBE, itemKeys)
            : queryExecutor.executePartitioned(BigtableTableName.YOUTUBE, itemKeys, List.of(DEMOGRAPHICS));
        return rows
            .thenApply(rowStream -> rowStream.map(row -> rowReader.readRow(YouTubeVideoStats.class, row)));
    }

}