package io.delphiplatform.api.v3.bigtable.processing;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;

import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.v3.bigtable.processing.groupbykey.YouTubeSummaryGroupByKey;
import io.delphiplatform.api.v3.model.video.GroupByVideo;
import io.delphiplatform.api.v3.model.video.YouTubeSummaryItem;
import io.delphiplatform.api.v3.view.util.Params;

@Service
public class YouTubeSummariesAggregator {

    private final YouTubeSummariesMergingService youTubeSummariesMergingService;

    public YouTubeSummariesAggregator(YouTubeSummariesMergingService youTubeSummariesMergingService) {
        this.youTubeSummariesMergingService = youTubeSummariesMergingService;
    }

    public Stream<YouTubeSummaryItem> joinWithIsrc(
        Stream<YouTubeSummaryItem> summaries,
        Map<String, ImmutableSet<String>> inputVideoIdsToIsrcs
    ) {
        ImmutableMap<String, ImmutableSet<String>> videoIdsToIsrcs = ImmutableMap.copyOf(inputVideoIdsToIsrcs);
        return summaries.flatMap(s -> {
            ImmutableSet<String> videoIsrcs = videoIdsToIsrcs.get(s.getDimensions().getVideoId());
            return videoIsrcs.stream().map(isrc -> {
                YouTubeSummaryItem copied = youTubeSummariesMergingService.copy(s);
                copied.getDimensions().isrc(isrc);
                return copied;
            });
        });
    }

    public Stream<YouTubeSummaryItem> groupByAggregate(Stream<YouTubeSummaryItem> streams, Params params) {
        Set<GroupByFieldEnum> groupByFields = getGroupByFields(params);
        return groupByAggregate(streams, groupByFields);
    }

    private Set<GroupByFieldEnum> getGroupByFields(Params params) {
        Set<GroupByFieldEnum> groupByFields = Sets.newHashSet(GroupByFieldEnum.COUNTRY_CODE);

        if (CollectionUtils.contains(params.getGroupByFieldsVideo(), GroupByVideo.ISRC)) {
            groupByFields.add(GroupByFieldEnum.ISRC);
        }
        if (CollectionUtils.contains(params.getGroupByFieldsVideo(), GroupByVideo.VIDEO_ID)) {
            groupByFields.add(GroupByFieldEnum.VIDEO_ID);
        }

        if (CollectionUtils.contains(params.getGroupByFieldsVideo(), GroupByVideo.DATE)) {
            groupByFields.add(GroupByFieldEnum.DATE);
        }

        if (CollectionUtils.isNotEmpty(params.getProjectNumbers())) {
            groupByFields.add(GroupByFieldEnum.PROJECT_NUMBER);
        }

        if (CollectionUtils.isNotEmpty(params.getProductFamilyIds())) {
            groupByFields.add(GroupByFieldEnum.PRODUCT_FAMILY_ID);
        }

        return groupByFields;
    }

    private Stream<YouTubeSummaryItem> groupByAggregate(Stream<YouTubeSummaryItem> summaries,
        Set<GroupByFieldEnum> groupByFields) {
        return summaries
            .collect(Collectors.groupingBy(s -> new YouTubeSummaryGroupByKey(s, groupByFields)))
            .entrySet().stream()
            .map(entry -> {
                YouTubeSummaryGroupByKey key = entry.getKey();
                List<YouTubeSummaryItem> group = entry.getValue();
                YouTubeSummaryItem sum = new YouTubeSummaryItem();

                aggregateGroup(group, sum);

                sum.getDimensions().countryCode(key.getCountryCode());
                key.populateSummary(sum);

                return sum;
            });
    }

    private void aggregateGroup(List<YouTubeSummaryItem> group, YouTubeSummaryItem sum) {
        sum.dimensions(youTubeSummariesMergingService.mergeDimensions(group));
        sum.metrics(youTubeSummariesMergingService.mergeMetrics(group));
    }

}