package io.delphiplatform.api.v3.bigtable;

import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.transaction.NotSupportedException;

import io.delphiplatform.api.extapi.SpotifyApiService;
import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.util.DateUtils;
import io.delphiplatform.api.util.model.ModelUtils;
import io.delphiplatform.api.v3.bigtable.entity.PlaylistPublicPosition;
import io.delphiplatform.api.v3.bigtable.processing.PaginationService;
import io.delphiplatform.api.v3.bigtable.processing.PlaylistTrackCurrentPositionSortingService;
import io.delphiplatform.api.v3.bigtable.processing.TrackPositionConverter;
import io.delphiplatform.api.v3.bigtable.processing.TrackPositionPublicAggregator;
import io.delphiplatform.api.v3.bigtable.reader.BigtableEntityReader.EntityReadingParams;
import io.delphiplatform.api.v3.constant.DspConstants;
import io.delphiplatform.api.v3.model.IncludeTracklistPlaylist;
import io.delphiplatform.api.v3.model.PlaylistDatesAwareModel;
import io.delphiplatform.api.v3.model.PublicPlaylist;
import io.delphiplatform.api.v3.model.trackposition.PlaylistsTrackPositionsCurrentTracklist;
import io.delphiplatform.api.v3.model.trackposition.PublicPlaylistTrackPosition;
import io.delphiplatform.api.util.model.PlaylistIdIsrcKey;
import io.delphiplatform.api.v3.rdb.service.dto.PublicPlaylistLookupFilters;
import io.delphiplatform.api.v3.rdb.service.playlist_public.PlaylistPublicService;
import io.delphiplatform.api.v3.view.util.Params;

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

@Service
public class BigtableTrackPositionCurrentTracklistPublicService {

    private static final int LAST_14_DAYS_DATE_OFFSET = 13;

    private final BigtableTrackPositionPublicQueryExecutor trackPositionPublicQueryExecutor;
    private final PaginationService paginationService;

    private final TrackPositionConverter trackPositionConverter;
    private final TrackPositionPublicAggregator trackPositionPublicAggregator;
    private final PlaylistPublicService playlistPublicService;
    private final BigtableTrackPositionPublicManager bigtableTrackPositionPublicManager;
    private final BigtableTrackPositionDataEnrichmentPublicService dataEnrichmentPublicService;
    private final PlaylistTrackCurrentPositionSortingService playlistTrackCurrentPositionSortingService;
    private final SpotifyApiService spotifyApiService;

    public BigtableTrackPositionCurrentTracklistPublicService(
        BigtableTrackPositionPublicQueryExecutor trackPositionPublicQueryExecutor,
        PaginationService paginationService,
        TrackPositionConverter trackPositionConverter,
        TrackPositionPublicAggregator trackPositionPublicAggregator,
        PlaylistPublicService playlistPublicService,
        BigtableTrackPositionPublicManager bigtableTrackPositionPublicManager,
        BigtableTrackPositionDataEnrichmentPublicService dataEnrichmentPublicService,
        PlaylistTrackCurrentPositionSortingService playlistTrackCurrentPositionSortingService,
        SpotifyApiService spotifyApiService
    ) {
        this.trackPositionPublicQueryExecutor = trackPositionPublicQueryExecutor;
        this.paginationService = paginationService;
        this.trackPositionConverter = trackPositionConverter;
        this.trackPositionPublicAggregator = trackPositionPublicAggregator;
        this.playlistPublicService = playlistPublicService;
        this.bigtableTrackPositionPublicManager = bigtableTrackPositionPublicManager;
        this.dataEnrichmentPublicService = dataEnrichmentPublicService;
        this.playlistTrackCurrentPositionSortingService = playlistTrackCurrentPositionSortingService;
        this.spotifyApiService = spotifyApiService;
    }

    @Transactional(readOnly = true)
    public CompletableFuture<List<PlaylistsTrackPositionsCurrentTracklist>> getPlaylistCurrentTrackPositions(Params params)
        throws NotSupportedException {
        List<String> requestedPlaylistIds = ModelUtils.filterPlaylistsByDsp(params.getPlaylistId(), params.getDsp());

        boolean noPlaylistIdsAfterDspFiltering = CollectionUtils.isEmpty(requestedPlaylistIds);

        if (noPlaylistIdsAfterDspFiltering) {
            return CompletableFuture.completedFuture(Collections.emptyList());
        }

        Params paramsProcessed = params.toBuilder()
            .startDate(DateUtils.getCurrentDate().minusDays(1))
            .endDate(DateUtils.getCurrentDate())
            .playlistId(requestedPlaylistIds)
            .build();

        List<PublicPlaylist> fetchedPlaylists = playlistPublicService.findAllByPlaylistIdFiltered(
            paramsProcessed.getPlaylistId(),
            PublicPlaylistLookupFilters.builder()
                .countryCodes(paramsProcessed.getCountryCode())
                .includeRemoved(false)
                .build()
        );

        final Predicate<PublicPlaylist> spotifyGeneralPlaylistPredicate = p -> !Boolean.TRUE.equals(p.isPersonalised()) &&
            DspConstants.SPOTIFY.equals(p.getDsp().getDspId());
        final Predicate<PublicPlaylist> playlistToRequestFromSpotifyPredicate = spotifyGeneralPlaylistPredicate
            .or(p -> !p.isSaveTrackList());

        List<String> playlistToRequestFromSpotify = getPlaylistToRequestFromSpotify(fetchedPlaylists, playlistToRequestFromSpotifyPredicate,
            paramsProcessed.getPlaylistId());
        List<String> otherPlaylist = fetchedPlaylists.stream()
            .filter(playlistToRequestFromSpotifyPredicate.negate())
            .map(PublicPlaylist::getPlaylistId)
            .collect(Collectors.toList());

        List<CompletableFuture<Stream<PublicPlaylistTrackPosition>>> completableFutures = Collections.synchronizedList(new ArrayList<>());
        if (!playlistToRequestFromSpotify.isEmpty()) {
            boolean includeTrackInfo = CollectionUtils.contains(paramsProcessed.getInclude(), IncludeTracklistPlaylist.TRACK_INFO.getValue());
            completableFutures.add(
                CompletableFuture.supplyAsync(
                        () -> spotifyApiService.getPlaylistsItemsAsTrackPositions(playlistToRequestFromSpotify, includeTrackInfo))
                    .thenApply(models -> enrichSpotifyApiTrackPositions(models, playlistToRequestFromSpotify))
            );
        }
        if (!otherPlaylist.isEmpty()) {
            completableFutures.add(
                trackPositionPublicQueryExecutor.getAllPublicPlaylistTrackPositions(
                        paramsProcessed.toBuilder()
                            .playlistId(otherPlaylist)
                            .build(),
                        EnumSet.of(EntityReadingParams.EXPAND_ISRC_BY_TRACK_CURRENT_POSITION))
                    .thenApply(models -> ModelUtils.filterForLatestDate(models, PlaylistPublicPosition::getPlaylistId,
                        PlaylistPublicPosition::getDate))
                    .thenApply(trackPositionConverter::convertPublicPlaylistPositions)
            );
        }

        return CompletableFuture.allOf(completableFutures.toArray(CompletableFuture[]::new))
            .thenApply(ignored -> completableFutures.stream().flatMap(CompletableFuture::join))
            .thenApply(trackPositionPublicAggregator::groupBy)
            .thenApply(models -> filterByCountryCodes(models, paramsProcessed.getCountryCode()))
            .thenApply(models -> dataEnrichmentPublicService.populatePlaylistStreamsForXDaysPeriods(models, paramsProcessed))
            .thenApply(trackPositionConverter::convertPlaylistPositionsToCurrentTracklist)
            .thenApply(dataEnrichmentPublicService::populateCurrentTrackPositionsMetrics)
            .thenApply(models -> bigtableTrackPositionPublicManager.populatePlaylistDates(models, paramsProcessed))
            .thenApply(models -> bigtableTrackPositionPublicManager.populateNumTracks(models, paramsProcessed,
                new HashSet<>(playlistToRequestFromSpotify)))
            .thenApply(models -> dataEnrichmentPublicService
                .populatePlaylistTrackInfoForCurrentTrackPositions(models, playlistToRequestFromSpotify, paramsProcessed))
            .thenApply(models -> playlistTrackCurrentPositionSortingService
                .sortCurrentPlaylistPositions(models, paramsProcessed.getSortList()))
            .thenApply(models -> paginationService.getPage(models, paramsProcessed))
            .thenApply(models -> models.collect(Collectors.toList()));
    }

    private List<String> getPlaylistToRequestFromSpotify(List<PublicPlaylist> playlists,
        Predicate<PublicPlaylist> playlistPredicate,
        List<String> requestedPlaylistIds) {

        Set<String> fetchedPlaylistsIds = playlists.stream()
            .map(PublicPlaylist::getPlaylistId)
            .collect(Collectors.toSet());
        List<String> notFetchedSpotifyPlaylistIds = requestedPlaylistIds.stream()
            .filter(p -> !fetchedPlaylistsIds.contains(p) && ModelUtils.isPlaylistOfDsp(p, SPOTIFY))
            .collect(Collectors.toList());

        return Stream.of(notFetchedSpotifyPlaylistIds,
                playlists.stream()
                    .filter(playlistPredicate)
                    .map(PublicPlaylist::getPlaylistId)
                    .collect(Collectors.toList()))
            .flatMap(Collection::stream)
            .collect(Collectors.toList());
    }

    private Stream<PublicPlaylistTrackPosition> enrichSpotifyApiTrackPositions(
        Stream<PublicPlaylistTrackPosition> models,
        List<String> spotifyPlaylist) {

        if (CollectionUtils.isEmpty(spotifyPlaylist)) {
            return models;
        }

        Params params = Params.builder()
            .startDate(DateUtils.getCurrentDate().minusDays(LAST_14_DAYS_DATE_OFFSET))
            .endDate(DateUtils.getCurrentDate())
            .playlistId(spotifyPlaylist)
            .build();

        try {
            Map<PlaylistIdIsrcKey, List<PlaylistPublicPosition>> isrcPlaylistPositionsFromBT = trackPositionPublicQueryExecutor
                .getAllPublicPlaylistTrackPositions(params).join()
                .collect(Collectors.groupingBy(tp -> ModelUtils.composePlaylistIsrcKeyFromValues(tp.getPlaylistId(), tp.getIsrc())));

            List<PublicPlaylistTrackPosition> trackPositionsFromSpotifyApi = models
                .sorted(Comparator.comparing(PublicPlaylistTrackPosition::getCurrent))
                .collect(Collectors.toList());

            // for multiple the same-ISRC-per-playlist process only the top one
            Map<PlaylistIdIsrcKey, PublicPlaylistTrackPosition> processedIsrcPlaylistPositions = new HashMap<>();
            trackPositionsFromSpotifyApi.forEach(spotifyPosition -> {
                PlaylistIdIsrcKey isrcForPlaylist = ModelUtils.composePlaylistIsrcKeyFromValues(spotifyPosition.getPlaylistId(),
                    spotifyPosition.getIsrc());

                //if we already processed a (top) position for this isrc/playlist - just copy its field values into this position
                if (processedIsrcPlaylistPositions.containsKey(isrcForPlaylist)) {
                    PublicPlaylistTrackPosition topPosition = processedIsrcPlaylistPositions.get(isrcForPlaylist);

                    spotifyPosition.setPrevious(topPosition.getPrevious());
                    spotifyPosition.setPreviousPositionDateTime(topPosition.getPreviousPositionDateTime());

                    spotifyPosition.setPreviousPositionChange14Days(topPosition.getPreviousPositionChange14Days());
                    spotifyPosition.setLastDateChange14Days(topPosition.getLastDateChange14Days());
                    spotifyPosition.setTrendChange14Days(topPosition.getTrendChange14Days());

                    spotifyPosition.setEarliestPosition(topPosition.getEarliestPosition());

                    return;
                }

                List<PlaylistPublicPosition> playlistPositionsFromBT = Optional.ofNullable(isrcPlaylistPositionsFromBT.get(isrcForPlaylist))
                    .orElse(Collections.emptyList());
                if (playlistPositionsFromBT.isEmpty()) {
                    return;
                }

                playlistPositionsFromBT.sort(Comparator.comparing(PlaylistPublicPosition::getDate).reversed());
                PlaylistPublicPosition latestPositionFromBT = playlistPositionsFromBT.get(0);

                LocalDateTime previousPositionDateTime;
                Long previousPosition;

                //Determine item's 'previous' (position) and 'previous position date' - if possible just use data from BT
                //but if BT data is outdated (not yet loaded for today or inside-day update is not yet received)
                // - use prev. available day's 'current' as previous

                //Spotify API item's 'current' (position) is in priority as freshest "today's position"
                //we compare it to 'latest' found BT item's 'current'
                //if 'latest' BT item's 'current' is equal to Spotify API - means BT item state is actual, and we can just use its 'previous'
                // (or - if item's report_date < today - BT data is behind Spotify API but doesn't matter in this case
                //      since 'previous' still points to latest 'different' position, and we can use it)
                if (spotifyPosition.getCurrent().equals(latestPositionFromBT.getCurrent())) {
                    //latest BT item's 'current' is fresh, so 'previous' is also fresh (even if it is null)
                    previousPosition = latestPositionFromBT.getPrevious();
                    previousPositionDateTime = latestPositionFromBT.getPreviousPositionDateTime();
                } else {
                    //latest BT item's 'current' is outdated (so either this item has report_date < today
                    //  or just hasn't received latest inside-day update yet)

                    //if BT item's report_date < today - just means its 'current' will be set as 'previous', once next record
                    //  with report_date = today is eventually created. So we can use it as 'previous' right away
                    if (latestPositionFromBT.getDate().isBefore(DateUtils.getCurrentDate())) {
                        previousPosition = latestPositionFromBT.getCurrent();
                        previousPositionDateTime = latestPositionFromBT.getDate().atStartOfDay();
                    } else {
                        //if BT item's report_date == today, but it doesn't yet know about fresh 'current' change - means its 'previous'
                        //value also may be outdated. So we have to look for 'latest-1' item (e.g. yesterday's) and use its 'current'
                        //  as 'previous' - to maintain per-day granularity of 'previous' value.
                        Optional<PlaylistPublicPosition> secondLatestPositionFromBTOpt = playlistPositionsFromBT.size() > 1
                            ? Optional.of(playlistPositionsFromBT.get(1)) //sorted, so this is 'second-latest' BT item
                            : Optional.empty();

                        if (secondLatestPositionFromBTOpt.isPresent()) {
                            PlaylistPublicPosition secondLatestPositionFromBT = secondLatestPositionFromBTOpt.get();

                            previousPosition = secondLatestPositionFromBT.getCurrent();
                            previousPositionDateTime = secondLatestPositionFromBT.getDate().atStartOfDay();
                        } else {
                            //in case if no 'latest-1' item found:
                            // - if this item existed in playlist before    - means it hasn't been changed during 14 days period,
                            //   and we don't need to return 'previous'-based fields
                            // - if it appeared today for the 1st time ever - just return empty 'previous'-based fields

                            previousPosition = null;
                            previousPositionDateTime = null;
                        }
                    }
                }

                if (previousPositionDateTime != null) {
                    spotifyPosition.setPrevious(previousPosition);
                    spotifyPosition.setPreviousPositionDateTime(previousPositionDateTime);

                    LocalDate previousPositionDate = previousPositionDateTime.toLocalDate();

                    if (previousPositionDate.isAfter(DateUtils.getCurrentDate().minusDays(LAST_14_DAYS_DATE_OFFSET))) {
                        spotifyPosition.setPreviousPositionChange14Days(spotifyPosition.getPrevious());
                        spotifyPosition.setLastDateChange14Days(previousPositionDate);
                        spotifyPosition.setTrendChange14Days(ModelUtils
                            .calculatePlaylistTrackPositionTrend(spotifyPosition.getPrevious(), spotifyPosition.getCurrent()));
                    }
                }

                spotifyPosition.setEarliestPosition(latestPositionFromBT.getEarliestPosition());

                processedIsrcPlaylistPositions.put(isrcForPlaylist, spotifyPosition);
            });

            return trackPositionsFromSpotifyApi.stream();
        } catch (NotSupportedException e) {
            throw new RuntimeException("Cannot execute BigTable query.", e);
        }
    }

    private <P extends PlaylistDatesAwareModel> Stream<P> filterByCountryCodes(Stream<P> models, Set<String> countryCodes) {
        if (CollectionUtils.isEmpty(countryCodes)) {
            return models;
        }

        return models
            .filter(p -> countryCodes.contains(p.getCountryCode()));
    }
}