package io.delphiplatform.api.v3.bigtable;

import org.springframework.stereotype.Service;

import java.math.RoundingMode;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
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.util.model.ModelUtils;
import io.delphiplatform.api.v3.constant.ApplicationConstants;
import io.delphiplatform.api.v3.constant.DspConstants;
import io.delphiplatform.api.v3.model.AggBy;
import io.delphiplatform.api.v3.model.GroupByField;
import io.delphiplatform.api.v3.model.IncludeTracklistPlaylist;
import io.delphiplatform.api.v3.model.Playlist;
import io.delphiplatform.api.v3.model.PlaylistableModel;
import io.delphiplatform.api.v3.model.StreamModel;
import io.delphiplatform.api.v3.model.SubsetParam;
import io.delphiplatform.api.v3.model.TrackAwareModel;
import io.delphiplatform.api.v3.model.TrackSimple;
import io.delphiplatform.api.v3.model.amazon.AmazonTrackShortInfo;
import io.delphiplatform.api.v3.model.amazon.StreamingFor1DaysInfo;
import io.delphiplatform.api.v3.model.amazon.StreamingFor7DaysInfo;
import io.delphiplatform.api.v3.model.trackposition.PlaylistTrackPosition;
import io.delphiplatform.api.v3.rdb.entity.ChartmetricToAmazonPlaylistIdMappingProjection;
import io.delphiplatform.api.v3.rdb.service.PlaylistManager;
import io.delphiplatform.api.v3.rdb.service.PlaylistService;
import io.delphiplatform.api.v3.rdb.service.TrackService;
import io.delphiplatform.api.v3.rdb.service.amazon.AmazonPlaylistTrackService;
import io.delphiplatform.api.v3.view.util.Params;

@Deprecated
@Service
public class BigtableTrackPositionDataEnrichmentService {

    private static final int TRACK_POS_STREAMS_FOR_PERIOD_OFFSET_DAYS = 7;
    private static final int TRACK_POS_STREAMS_FOR_PERIOD_OFFSET_DAYS_LAG = 2;

    private final PlaylistService playlistService;
    private final PlaylistManager playlistManager;
    private final BigtableStreamsService bigtableStreamsService;
    private final TrackService trackService;
    private final AmazonPlaylistTrackService amazonPlaylistTrackService;

    public BigtableTrackPositionDataEnrichmentService(
        PlaylistService playlistService,
        PlaylistManager playlistManager,
        BigtableStreamsService bigtableStreamsService,
        TrackService trackService,
        AmazonPlaylistTrackService amazonPlaylistTrackService
    ) {
        this.playlistService = playlistService;
        this.playlistManager = playlistManager;
        this.bigtableStreamsService = bigtableStreamsService;
        this.trackService = trackService;
        this.amazonPlaylistTrackService = amazonPlaylistTrackService;
    }

    public Stream<PlaylistTrackPosition> populatePlaylistStreamsForXDaysPeriods(Stream<PlaylistTrackPosition> models, Params params) {
        if (!CollectionUtils.contains(params.getInclude(), IncludeTracklistPlaylist.STREAMS_FOR_PERIOD.getValue())) {
            return models;
        }

        List<PlaylistTrackPosition> positions = models.collect(Collectors.toList());

        if (positions.stream().anyMatch(position -> !DspConstants.AMAZON.equals(position.getDsp()))) {
            throw new IllegalArgumentException("only Amazon is supported in this deprecated method");
        }

        if (!CollectionUtils.contains(params.getInclude(), IncludeTracklistPlaylist.STREAMS_FOR_PERIOD.getValue())
            || positions.isEmpty()) {
            return positions.stream();
        }

        Set<String> isrcs = positions.stream()
            .map(PlaylistTrackPosition::getIsrc)
            .collect(Collectors.toSet());

        Set<String> playlistIdsAmazon = new HashSet<>();

        //mapping between Chartmetric Amazon playlist id with country and Amazon playlist id
        Map<String, String> cmAmazonPlaylistIdToAmazonPlaylistId = new HashMap<>();

        Set<String> amazonCmPlaylistIds = positions.stream()
            .filter(position -> DspConstants.AMAZON.equals(position.getDsp()))
            .filter(position -> CollectionUtils.isNotEmpty(position.getCountryCode()))
            .map(PlaylistTrackPosition::getPlaylistId)
            .collect(Collectors.toSet());

        if (!amazonCmPlaylistIds.isEmpty()) {
            cmAmazonPlaylistIdToAmazonPlaylistId.putAll(
                playlistService.getChartmetricToAmazonPlaylistIdMapping(amazonCmPlaylistIds).stream()
                    .collect(Collectors.toMap(
                        ChartmetricToAmazonPlaylistIdMappingProjection::getCmAmazonPlaylistIdWithCountry,
                        projection -> ModelUtils.getPlaylistIdWithDspPrefix(DspConstants.AMAZON, projection.getAmazonPlaylistId())
                    ))
            );

            playlistIdsAmazon.addAll(
                cmAmazonPlaylistIdToAmazonPlaylistId.values()
            );
        }

        LocalDate endDate = params.getEndDate();
        LocalDate startDate = endDate.minusDays(TRACK_POS_STREAMS_FOR_PERIOD_OFFSET_DAYS + TRACK_POS_STREAMS_FOR_PERIOD_OFFSET_DAYS_LAG);
        List<String> dsps = CollectionUtils.isEmpty(params.getDsp())
            ? List.of(DspConstants.APPLE, DspConstants.SPOTIFY, DspConstants.AMAZON)
            : params.getDsp();

        if (playlistIdsAmazon.isEmpty()) {
            return positions.stream();
        }

        List<String> playlistIds = new ArrayList<>(playlistIdsAmazon);

        Params streamsGroupedByPlaylistAndIsrcAndDateParams = Params.builder()
            .startDate(startDate)
            .endDate(endDate)
            .isrc(new ArrayList<>(isrcs))
            .playlistId(playlistIds)
            .countryCode(params.getStreamsCountryCode())
            .dsp(dsps)
            .aggBy(AggBy.ISRC)
            .subset(SubsetParam.PLAYLISTS)
            .groupByFields(Set.of(GroupByField.DATE, GroupByField.COUNTRY))
            .limit(Integer.MAX_VALUE)
            .build();

        Params streamsGroupedByPlaylistAndDateParams = Params.builder()
            .startDate(startDate)
            .endDate(endDate)
            .playlistId(playlistIds)
            .countryCode(params.getStreamsCountryCode())
            .dsp(dsps)
            .subset(SubsetParam.PLAYLISTS)
            .groupByFields(Set.of(GroupByField.DATE, GroupByField.COUNTRY))
            .limit(Integer.MAX_VALUE)
            .build();

        return findAndPopulateStreamsForXDaysPeriods(
            streamsGroupedByPlaylistAndIsrcAndDateParams,
            streamsGroupedByPlaylistAndDateParams,
            positions,
            cmAmazonPlaylistIdToAmazonPlaylistId
        );
    }

    private Stream<PlaylistTrackPosition> findAndPopulateStreamsForXDaysPeriods(
        Params streamsGroupedByPlaylistAndIsrcAndDateParams,
        Params streamsGroupedByPlaylistAndDateParams,
        List<PlaylistTrackPosition> positions,
        Map<String, String> cmAmazonPlaylistIdToAmazonPlaylistId
    ) {
        try {
            CompletableFuture<List<StreamModel>> streamsGroupedByPlaylistAndIsrcAndDateFuture =
                bigtableStreamsService.getRows(streamsGroupedByPlaylistAndIsrcAndDateParams);

            CompletableFuture<List<StreamModel>> streamsGroupedByPlaylistAndDateFuture =
                bigtableStreamsService.getRows(streamsGroupedByPlaylistAndDateParams);

            CompletableFuture.allOf(streamsGroupedByPlaylistAndIsrcAndDateFuture,
                streamsGroupedByPlaylistAndDateFuture).get();

            Comparator<StreamModel> reversedDateComparator = Comparator.comparing(StreamModel::getDate).reversed();
            List<StreamModel> streamsGroupedByPlaylistAndIsrcAndDate = streamsGroupedByPlaylistAndIsrcAndDateFuture.get()
                .stream().sorted(reversedDateComparator).collect(Collectors.toList());

            List<StreamModel> streamsGroupedByPlaylistAndDate = streamsGroupedByPlaylistAndDateFuture.get()
                .stream().sorted(reversedDateComparator).collect(Collectors.toList());

            LocalDate latestAvailableDate = CollectionUtils.isNotEmpty(streamsGroupedByPlaylistAndIsrcAndDate) ?
                streamsGroupedByPlaylistAndIsrcAndDate.get(0).getDate() : null;
            LocalDate startDateFor7DayNotIncluding = latestAvailableDate != null
                ? latestAvailableDate.minusDays(TRACK_POS_STREAMS_FOR_PERIOD_OFFSET_DAYS)
                : null;

            Map<String, Long> streamsGroupedByPlaylistAndIsrcFor1Day = streamsGroupedByPlaylistAndIsrcAndDate
                .stream()
                .takeWhile(streamModel -> streamModel.getDate().equals(latestAvailableDate))
                .collect(Collectors.toMap(
                    streamModel -> ModelUtils.getPlaylistIdWithIsrc(streamModel.getPlaylistId(), streamModel.getIsrc()),
                    StreamModel::getStreams,
                    //if >1 stream countries requested/found for single playlistId/isrc at 'latestAvailableDate' - sum streams
                    (s1, s2) -> s1 + s2
                ));

            Map<String, Long> streamsGroupedByPlaylistAndIsrc7Day = streamsGroupedByPlaylistAndIsrcAndDate
                .stream()
                .takeWhile(streamModel -> startDateFor7DayNotIncluding != null
                    && streamModel.getDate().isAfter(startDateFor7DayNotIncluding))
                .collect(Collectors.toMap(
                    streamModel -> ModelUtils.getPlaylistIdWithIsrc(streamModel.getPlaylistId(), streamModel.getIsrc()),
                    StreamModel::getStreams,
                    (s1, s2) -> s1 + s2
                ));

            Map<String, Long> streamsGroupedByPlaylist7Day = streamsGroupedByPlaylistAndDate
                .stream()
                .takeWhile(streamModel -> startDateFor7DayNotIncluding != null
                    && streamModel.getDate().isAfter(startDateFor7DayNotIncluding))
                .collect(Collectors.toMap(
                    StreamModel::getPlaylistId,
                    StreamModel::getStreams,
                    (s1, s2) -> s1 + s2
                ));

            return positions.stream()
                .peek(position -> {
                    String playlistIdToQueryBTForStreams;

                    //to query BT amazon_music table - resolve 'amazon playlist id' by 'chartmetric amazon playlist id' which is used in position class
                    playlistIdToQueryBTForStreams =
                        cmAmazonPlaylistIdToAmazonPlaylistId.get(position.getPlaylistId());

                    if (playlistIdToQueryBTForStreams != null) {
                        String playlistIdAndIsrcKey = ModelUtils.getPlaylistIdWithIsrc(playlistIdToQueryBTForStreams, position.getIsrc());
                        Long isrcInPlaylistStreams = streamsGroupedByPlaylistAndIsrc7Day.get(playlistIdAndIsrcKey);
                        Long playlistStreams = streamsGroupedByPlaylist7Day.get(playlistIdToQueryBTForStreams);

                        if (isrcInPlaylistStreams != null || playlistStreams != null) {
                            position.setStreamingFor7Days(
                                StreamingFor7DaysInfo.builder()
                                    .isrcInPlaylist(isrcInPlaylistStreams)
                                    .playlist(playlistStreams)
                                    .build());
                        }

                        Long isrcInPlaylistStreams1Day = streamsGroupedByPlaylistAndIsrcFor1Day.get(playlistIdAndIsrcKey);
                        if (isrcInPlaylistStreams1Day != null) {
                            position.setStreamingFor1Days(
                                StreamingFor1DaysInfo.builder()
                                    .isrcInPlaylist(isrcInPlaylistStreams1Day)
                                    .build());
                        }
                    }
                });

        } catch (NotSupportedException | InterruptedException | ExecutionException e) {
            throw new RuntimeException("Cannot get future result for BigTable query.", e);
        }
    }

    public <T extends PlaylistableModel> Stream<T> populatePlaylists(Stream<T> models, Params params) {
        if (CollectionUtils.contains(params.getInclude(), IncludeTracklistPlaylist.PLAYLISTS.getValue())) {
            List<T> positions = models.collect(Collectors.toList());

            if (positions.isEmpty()) {
                return Stream.empty();
            }

            return playlistManager.populatePlaylists(positions)
                .stream()
                .filter(p -> p.getPlaylist() != null);
        }
        return models;
    }

    public <T extends PlaylistableModel> Stream<T> populateAmazonPlaylistsWithDspPlaylistId(
        Stream<T> models, Params params
    ) {
        if (CollectionUtils.contains(params.getInclude(), IncludeTracklistPlaylist.PLAYLISTS.getValue())) {
            List<T> playlistableModels = models.collect(Collectors.toList());

            if (playlistableModels.isEmpty()) {
                return Stream.empty();
            }

            List<Playlist> amazonPlaylists = playlistableModels.stream()
                .map(PlaylistableModel::getPlaylist)
                .filter(Objects::nonNull)
                .filter(p -> DspConstants.AMAZON.equals(p.getDsp().getDspId()))
                .collect(Collectors.toList());

            playlistService.populatePlaylistsWithAmazonPlaylistId(amazonPlaylists);

            return playlistableModels.stream();
        }

        return models;
    }

    public <T extends TrackAwareModel> Stream<T> populatePlaylistPositionTracks(Stream<T> models, Params params) {
        if (CollectionUtils.contains(params.getInclude(), IncludeTracklistPlaylist.TRACK_INFO.getValue())) {
            List<T> trackAwareModels = models.collect(Collectors.toList());

            if (trackAwareModels.isEmpty()) {
                return Stream.empty();
            }

            Map<String, TrackSimple> trackModelsByIsrc = trackService.findMainProductSimpleTracksByIsrcs(
                trackAwareModels.stream()
                    .map(T::getIsrc)
                    .collect(Collectors.toSet())
            );

            trackAwareModels.forEach(position -> position.setTrack(trackModelsByIsrc.get(position.getIsrc())));

            return trackAwareModels.stream();
        } else {
            return models;
        }
    }

    public Stream<PlaylistTrackPosition> populatePlaylistPositionAmazonTracks(
        Stream<PlaylistTrackPosition> trackPositionsStream,
        Params params
    ) {
        if (CollectionUtils.contains(params.getInclude(), IncludeTracklistPlaylist.TRACK_INFO.getValue())) {
            List<PlaylistTrackPosition> positions = trackPositionsStream.collect(Collectors.toList());

            if (positions.isEmpty()) {
                return Stream.empty();
            }

            Set<PlaylistTrackPosition> amazonPositions = positions.stream()
                .filter(position -> DspConstants.AMAZON.equals(position.getDsp()))
                .collect(Collectors.toSet());

            if (!amazonPositions.isEmpty()) {
                Map<String, AmazonTrackShortInfo> amazonTracksById = amazonPlaylistTrackService.findById(
                        amazonPositions.stream()
                            .map(PlaylistTrackPosition::getAmazonTrackIdUnPrefixed)
                            .filter(Objects::nonNull)
                            .collect(Collectors.toSet())
                    )
                    .stream()
                    .collect(Collectors.toMap(AmazonTrackShortInfo::getAmazonTrackId, Function.identity()));

                amazonPositions.forEach(position -> position.setAmazonTrack(
                    amazonTracksById.get(position.getAmazonTrackIdUnPrefixed()))
                );
            }

            return positions.stream();
        } else {
            return trackPositionsStream;
        }
    }

    public Stream<PlaylistTrackPosition> populateNumWeeksOnPlaylist(Stream<PlaylistTrackPosition> trackPositionsStream) {
        return trackPositionsStream.peek(position -> {
            Long daysInPlaylist = position.getDaysInPlaylist();

            if (daysInPlaylist != null) {
                position.setNumWeeksOn(
                    ModelUtils.toBigDecimal(daysInPlaylist)
                        .divide(ModelUtils.toBigDecimal(ApplicationConstants.DAYS_IN_WEEK), RoundingMode.FLOOR)
                        .intValue()
                );
            }
        });
    }
}