package io.delphiplatform.api.v3.rdb.service;

import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.util.DateUtils;
import io.delphiplatform.api.v3.model.StreamModel;
import io.delphiplatform.api.v3.model.TrackStreamsInsightsItem;

import static io.delphiplatform.api.v3.constant.DspConstants.AMAZON;
import static io.delphiplatform.api.v3.constant.DspConstants.APPLE;
import static io.delphiplatform.api.v3.constant.DspConstants.SPOTIFY;

@Service
public class TrackStreamsInsightsAggregateService {

    private static final String ALL_DSPS = "all_dsps";

    public Map<String, List<TrackStreamsInsightsItem>> aggregate(List<StreamModel> models, List<String> requestedDsps) {
        boolean includeAllDspsPlusMergedStreams = CollectionUtils.isEmpty(requestedDsps);

        if (CollectionUtils.isEmpty(models)) {
            return emptyResponse(requestedDsps, includeAllDspsPlusMergedStreams);
        }

        List<StreamModel> processedModels;

        if (includeAllDspsPlusMergedStreams) {
            processedModels = CollectionUtils.merge2Lists(models, mergeStreamModelsOfAllDsps(models));
        } else {
            processedModels = models;
        }

        Map<String, Map<String, StreamModel>> peakStreamsByDspAndCountry = processedModels.stream()
            .collect(Collectors.groupingBy(StreamModel::getDsp,
                Collectors.toMap(StreamModel::getCountryCode,
                    Function.identity(),
                    this::chosePeakStreamsDate
                )));

        // top 10 StreamModel with max streams per dsp or for "all_dsps"
        Map<String, List<StreamModel>> topStreamsByDsp = peakStreamsByDspAndCountry.entrySet().stream()
            .collect(
                Collectors.toMap(Entry::getKey,
                    entry -> entry.getValue()
                        .values().stream()
                        .sorted(Comparator.comparing(StreamModel::getStreams).reversed()
                            .thenComparing(StreamModel::getCountryCode))
                        .collect(Collectors.toList())
                )
            );

        ensureResponseEntriesForAllRequestedDsps(topStreamsByDsp, includeAllDspsPlusMergedStreams, requestedDsps);

        Set<String> topStreamsCountries = topStreamsByDsp.entrySet().stream()
            .flatMap(entry -> entry.getValue().stream())
            .map(StreamModel::getCountryCode)
            .collect(Collectors.toSet());

        Map<String, Map<String, Long>> strikeWeeksByDspAndCountry = prepareStrikeWeeksCountByDspAndCountry(
            processedModels.stream()
                .filter(sm -> topStreamsCountries.contains(sm.getCountryCode()))
                .collect(Collectors.toList()),
            includeAllDspsPlusMergedStreams
        );

        return topStreamsByDsp.entrySet().stream()
            .sorted(Entry.comparingByKey())
            .collect(Collectors.toMap(
                Entry::getKey,
                entry -> entry.getValue().stream()
                    .map(v -> TrackStreamsInsightsItem.from(v, strikeWeeksByDspAndCountry.get(entry.getKey())))
                    .collect(Collectors.toList()),
                (m1, m2) -> m1,
                LinkedHashMap::new
            ));
    }

    /**
     * Merges models of different DSPs into one (grouping by country, date) and sets dsp="all_dsps")
     */
    private List<StreamModel> mergeStreamModelsOfAllDsps(List<StreamModel> models) {
        Map<String, Map<LocalDate, Long>> mergedStreamsByDateAndCountry = models.stream()
            .collect(Collectors.groupingBy(StreamModel::getCountryCode,
                Collectors.toMap(StreamModel::getDate,
                    StreamModel::getStreams,
                    Long::sum
                )));

        List<StreamModel> mergedStreamModels = mergedStreamsByDateAndCountry.entrySet().stream()
            .flatMap(countryEntry -> countryEntry.getValue().entrySet().stream()
                .map(dateEntry -> new StreamModel()
                    .mainDsp(ALL_DSPS)
                    .countryCode(countryEntry.getKey())
                    .date(dateEntry.getKey())
                    .streams(dateEntry.getValue())
                )
            ).collect(Collectors.toList());

        return mergedStreamModels;
    }

    /**
     * Calculates streaming 'strike weeks' count - for each country of each dsp
     * <p>
     * E.g. if track has 150 streams at week3 and 130 at week2 and 200 at week1 - (most recent) strike weeks are week3 and week2, so count
     * is 2
     *
     * @param models                - StreamModel's created during processing - may either be just a list of BT models or list of BT models
     *                              plus merged cross-dsp models (see mergeStreamModelsOfAllDsps())
     * @param mergedStreamsIncluded - true means that `models` list param contains additional merged cross-dsp models
     * @return map of {dsp: {country: weeksCount}}
     */
    private Map<String, Map<String, Long>> prepareStrikeWeeksCountByDspAndCountry(
        List<StreamModel> models,
        boolean mergedStreamsIncluded
    ) {
        //find a date of the current "chart week" start
        LocalDate currentChartWeekStartDate = DateUtils.getChartStartDate(DateUtils.getCurrentDate());

        List<StreamModel> modelsFiltered = models.stream()
            .filter(streamModel -> !streamModel.getDate().isAfter(currentChartWeekStartDate))
            .collect(Collectors.toList());

        //find a date of the latest "chart week" start, related to the most recent streaming date available per country and dsp
        Map<String, Map<String, LocalDate>> latestChartWeekStartDateByCountryAndDsp = modelsFiltered.stream()
            .filter(m -> !ALL_DSPS.equals(m.getDsp()))
            .collect(Collectors.groupingBy(StreamModel::getCountryCode,
                Collectors.toMap(StreamModel::getDsp,
                    StreamModel::getDate,
                    (d1, d2) -> DateUtils.getChartStartDate(d1.isAfter(d2) ? d1 : d2)
                )));

        //for merged cross-dsp streams - 'chart week start date' should exist for all DSPs: amazon, apple, spotify
        //so, take minDate among existing DSPs
        if (mergedStreamsIncluded) {
            latestChartWeekStartDateByCountryAndDsp.forEach((k, datesByDsp) -> {
                LocalDate minDate = datesByDsp.values().stream()
                    .min(LocalDate::compareTo)
                    .get();

                datesByDsp.put(ALL_DSPS, minDate);
            });
        }

        //group StreamModel's by dsp and country, and calculate number of "strike weeks" for each group
        return modelsFiltered.stream()
            .sorted(Comparator.comparing(StreamModel::getDate).reversed())
            .collect(Collectors.groupingBy(StreamModel::getDsp,
                Collectors.groupingBy(StreamModel::getCountryCode, Collectors.collectingAndThen(Collectors.toList(),
                    streamModels -> {
                        StreamModel anyModel = streamModels.iterator().next();

                        LocalDate latestChartWeekStartDate = latestChartWeekStartDateByCountryAndDsp
                            .get(anyModel.getCountryCode())
                            .get(anyModel.getDsp());

                        //iterate StreamModel's (sorted by date desc, so from latest to oldest)
                        //going from the latest date backwards, gather sum of all streams of each earlier week and store to weeksStreamsQueue
                        //while number of streams is decreasing for each next (so actually each "earlier") week
                        // - means track was getting more popular each week and had "strike weeks".
                        //once we reach the earliest available date OR streams number stops decreasing during iteration
                        // - streak ends, and we return number of weeks gathered so far.

                        List<Long> weeksStreamsQueue = new ArrayList<>();
                        Long streamsPerWeek = 0L;

                        for (StreamModel streamModel : streamModels) {
                            if (streamModel.getDate().isAfter(latestChartWeekStartDate)) {
                                continue;
                            }

                            long daysTillLatestDate = ChronoUnit.DAYS.between(streamModel.getDate(), latestChartWeekStartDate);
                            long weekIndex = daysTillLatestDate / 7;

                            if (weeksStreamsQueue.size() == weekIndex) {
                                streamsPerWeek += streamModel.getStreams();
                                continue;
                            }

                            if (weeksStreamsQueue.size() > 0) {
                                long streamsOfLastCheckedWeek = weeksStreamsQueue.get(weeksStreamsQueue.size() - 1);
                                //if last (so actually "more recent") week's streams are less than current - then streak ended
                                if (streamsOfLastCheckedWeek <= streamsPerWeek) {
                                    break;
                                }
                            }

                            weeksStreamsQueue.add(streamsPerWeek);
                            streamsPerWeek = streamModel.getStreams();
                        }
                        return Long.max(weeksStreamsQueue.size() - 1, 0);
                    }
                ))));
    }

    private void ensureResponseEntriesForAllRequestedDsps(
        Map<String, List<StreamModel>> topStreamsByDsp,
        boolean includeAllDspsPlusMergedStreams,
        List<String> requestedDsps
    ) {
        if (includeAllDspsPlusMergedStreams || requestedDsps.contains(SPOTIFY)) {
            topStreamsByDsp.computeIfAbsent(SPOTIFY, ignored -> new ArrayList<>());
        }

        if (includeAllDspsPlusMergedStreams || requestedDsps.contains(APPLE)) {
            topStreamsByDsp.computeIfAbsent(APPLE, ignored -> new ArrayList<>());
        }

        if (includeAllDspsPlusMergedStreams || requestedDsps.contains(AMAZON)) {
            topStreamsByDsp.computeIfAbsent(AMAZON, ignored -> new ArrayList<>());
        }
    }

    private Map<String, List<TrackStreamsInsightsItem>> emptyResponse(List<String> dsps, boolean includeAllDspsPlusMergedStreams) {
        Set<String> dspsAtResponse = includeAllDspsPlusMergedStreams ? Set.of(ALL_DSPS, SPOTIFY, APPLE, AMAZON) : Set.copyOf(dsps);
        return dspsAtResponse.stream()
            .collect(Collectors.toMap(dsp -> dsp, ignored -> Collections.emptyList()));
    }

    private StreamModel chosePeakStreamsDate(StreamModel sm1, StreamModel sm2) {
        if (sm1.getStreams().equals(sm2.getStreams())) {
            return sm1.getDate().isBefore(sm2.getDate()) ? sm1 : sm2;
        }

        return sm1.getStreams() > sm2.getStreams() ? sm1 : sm2;
    }
}