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

import org.openapitools.jackson.nullable.JsonNullable;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
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.StreamGroupByKey;
import io.delphiplatform.api.v3.model.AggBy;
import io.delphiplatform.api.v3.model.GroupByField;
import io.delphiplatform.api.v3.model.IncludeStreams;
import io.delphiplatform.api.v3.model.StreamModel;
import io.delphiplatform.api.v3.model.SubsetParam;
import io.delphiplatform.api.v3.model.spotify.SpotifySaves;
import io.delphiplatform.api.v3.view.util.Params;

@Service
public class StreamsAggregator {

    private final StreamsMergingService streamsMergingService;

    public StreamsAggregator(StreamsMergingService streamsMergingService) {
        this.streamsMergingService = streamsMergingService;
    }

    public Stream<StreamModel> groupByAggregate(Stream<StreamModel> streams,
        Params params) {
        List<GroupByFieldEnum> groupByFields = getGroupByFields(params);
        String dsp = params.getDsp() != null && params.getDsp().size() == 1 ? params.getDsp().get(0) : null;
        String playlistId =
            params.getPlaylistId() != null && params.getPlaylistId().size() == 1 ? params.getPlaylistId().get(0) : null;

        return groupByAggregate(streams, groupByFields, dsp, playlistId, params.getIncludeStreams());
    }

    private List<GroupByFieldEnum> getGroupByFields(Params params) {
        List<GroupByFieldEnum> groupByFields = new ArrayList<>();

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

        if (CollectionUtils.isNotEmpty(params.getArtistId())) {
            groupByFields.add(GroupByFieldEnum.ARTIST_ID);
        }

        if (CollectionUtils.isNotEmpty(params.getIsrc())
            || AggBy.ISRC.equals(params.getAggBy())) {
            groupByFields.add(GroupByFieldEnum.ISRC);
        }

        if (CollectionUtils.isNotEmpty(params.getTrackId())) {
            groupByFields.add(GroupByFieldEnum.TRACK_ID);
        }

        if (SubsetParam.PLAYLISTS.equals(params.getSubset())) {
            groupByFields.add(GroupByFieldEnum.PLAYLIST_ID);
        }

        if (CollectionUtils.contains(params.getGroupByFields(), GroupByField.DATE)) {
            groupByFields.add(GroupByFieldEnum.DATE);
        }

        if (CollectionUtils.contains(params.getGroupByFields(), GroupByField.SUB_DSP)) {
            groupByFields.add(GroupByFieldEnum.SUB_DSP);
        }

        if (CollectionUtils.contains(params.getGroupByFields(), GroupByField.COUNTRY)) {
            groupByFields.add(GroupByFieldEnum.COUNTRY_CODE);
        }

        return groupByFields;
    }

    private Stream<StreamModel> groupByAggregate(Stream<StreamModel> streams,
        List<GroupByFieldEnum> groupByFields, String dsp, String playlistId, Set<IncludeStreams> includeStreams) {
        return streams
            .collect(Collectors.groupingBy(s -> new StreamGroupByKey(s, groupByFields)))
            .entrySet().stream()
            .map(entry -> {
                StreamGroupByKey key = entry.getKey();
                List<StreamModel> group = entry.getValue();
                StreamModel sum = new StreamModel();

                if (dsp != null) {
                    sum.mainDsp(dsp);
                }
                if (playlistId != null) {
                    sum.playlistId(playlistId);
                }
                aggregateGroup(includeStreams, group, sum);

                sum.streams(group.stream()
                    .filter(s -> s.getStreams() != null)
                    .mapToLong(StreamModel::getStreams)
                    .sum());
                sum.setHasStationStreams(hasStationStreams(group));

                key.populateStream(sum);

                return sum;
            });
    }

    private void aggregateGroup(Set<IncludeStreams> includeStreams, List<StreamModel> group, StreamModel sum) {
        boolean isDemographicsIncluded = CollectionUtils.contains(includeStreams, IncludeStreams.DEMOGRAPHICS);
        boolean isAllIncluded = CollectionUtils.contains(includeStreams, IncludeStreams.ALL);
        boolean isSavesIncluded = CollectionUtils.contains(includeStreams, IncludeStreams.SAVES);
        boolean isSkipsIncluded = CollectionUtils.contains(includeStreams, IncludeStreams.SKIPS);
        boolean isDspStreamsInfo = CollectionUtils.contains(includeStreams, IncludeStreams.DSP_STREAMS_INFO);

        if (isAllIncluded) {
            sum.setAppleStreamsInfo(
                JsonNullable.of(streamsMergingService.mergeAppleStreamInfo(group).orElse(null)));
            sum.setAmazonStreamsInfo(
                JsonNullable.of(streamsMergingService.mergeAmazonStreamInfo(group).orElse(null)));
            sum.setSpotifyStreamsInfo(
                JsonNullable.of(streamsMergingService.mergeSpotifyStreamInfo(group).orElse(null)));
            sum.setAppleAgeBands(JsonNullable.of(streamsMergingService.mergeAppleAgeBand(group).orElse(null)));
            sum.setSpotifyAgeBands(JsonNullable.of(streamsMergingService.mergeSpotifyAgeBand(group).orElse(null)));
            sum.setGenders(JsonNullable.of(streamsMergingService.mergeGender(group).orElse(null)));
        } else {
            if (isDemographicsIncluded) {
                sum.setAppleStreamsInfo(JsonNullable.of(null));
                sum.setAmazonStreamsInfo(JsonNullable.of(null));
                sum.setSpotifyStreamsInfo(JsonNullable.of(null));
                sum.setAppleAgeBands(JsonNullable.of(streamsMergingService.mergeAppleAgeBand(group).orElse(null)));
                sum.setSpotifyAgeBands(JsonNullable.of(streamsMergingService.mergeSpotifyAgeBand(group).orElse(null)));
                sum.setGenders(JsonNullable.of(streamsMergingService.mergeGender(group).orElse(null)));
            }
            if (isDspStreamsInfo) {
                sum.setAppleStreamsInfo(streamsMergingService.mergeAppleStreamInfo(group));
                sum.setAmazonStreamsInfo(streamsMergingService.mergeAmazonStreamInfo(group));
                sum.setSpotifyStreamsInfo(streamsMergingService.mergeSpotifyStreamInfo(group));
            } else {
                if (isSavesIncluded) {
                    JsonNullable<SpotifySaves> saves = streamsMergingService.mergeSpotifySaves(group);
                    if (saves.isPresent()) {
                        sum.initSpotifyStreamsInfoIfNone().setSaves(saves);
                    }
                }
                if (isSkipsIncluded) {
                    JsonNullable<Long> spotifySkips = streamsMergingService.mergeSpotifySkips(group);
                    if (spotifySkips.isPresent()) {
                        sum.initSpotifyStreamsInfoIfNone().setSkips(spotifySkips);
                    }
                    JsonNullable<Long> appleSkips = streamsMergingService.mergeAppleSkips(group);
                    if (appleSkips.isPresent()) {
                        sum.initAppleStreamsInfoIfNone().setSkips(appleSkips);
                    }
                }
            }
        }
    }

    private Boolean hasStationStreams(List<StreamModel> group) {
        return group.stream()
            .filter(s -> Boolean.TRUE.equals(s.hasStationStreams()))
            .findAny()
            .map(s -> true)
            .orElse(
                group.stream()
                    .filter(s -> Boolean.FALSE.equals(s.hasStationStreams()))
                    .findAny()
                    .map(s -> false)
                    .orElse(null)
            );
    }
}