package io.delphiplatform.api.v3.bigtable;

import org.openapitools.jackson.nullable.JsonNullable;
import org.springframework.data.domain.Pageable;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.server.ResponseStatusException;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Comparator;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
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.DateUtils;
import io.delphiplatform.api.util.model.ModelUtils;
import io.delphiplatform.api.v3.bigtable.entity.PlaylistPublicDaily;
import io.delphiplatform.api.v3.bigtable.processing.PaginationService;
import io.delphiplatform.api.v3.bigtable.processing.TrackPositionConverter;
import io.delphiplatform.api.v3.bigtable.processing.TrackPositionPublicAggregator;
import io.delphiplatform.api.v3.bigtable.processing.TrackPositionSortingPublicService;
import io.delphiplatform.api.v3.bigtable.reader.BigtableEntityReader.EntityReadingParams;
import io.delphiplatform.api.v3.constant.CountryCodeConstant;
import io.delphiplatform.api.v3.constant.DspConstants;
import io.delphiplatform.api.v3.dto.ModelsListWithMetadataDto;
import io.delphiplatform.api.v3.model.PlaylistPublicPositionsDate;
import io.delphiplatform.api.v3.model.PlaylistSaveFlags;
import io.delphiplatform.api.v3.model.PublicPlaylist;
import io.delphiplatform.api.v3.model.trackposition.PublicPlaylistTrackPosition;
import io.delphiplatform.api.v3.model.trackposition.PublicPlaylistTrackPositionsMeta;
import io.delphiplatform.api.v3.model.video.ExpandTo;
import io.delphiplatform.api.v3.rdb.service.apple.music.AppleMusicPlaylistLifetimePublicService;
import io.delphiplatform.api.util.model.PlaylistIdCountryCodeNullableKey;
import io.delphiplatform.api.v3.rdb.service.dto.PublicPlaylistLookupFilters;
import io.delphiplatform.api.v3.rdb.service.playlist_public.PlaylistPublicService;
import io.delphiplatform.api.v3.rdb.service.spotify.SpotifyPlaylistLifetimePublicService;
import io.delphiplatform.api.v3.view.util.Params;

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

@Service
public class BigtableTrackPositionPublicService {

    private static final int DAYS_SHIFT_TO_SCAN_FOR_LATEST_DATA = 7;

    private final BigtableTrackPositionPublicQueryExecutor trackPositionPublicQueryExecutor;
    private final PaginationService paginationService;

    private final TrackPositionConverter trackPositionConverter;
    private final TrackPositionPublicAggregator trackPositionPublicAggregator;
    private final TrackPositionSortingPublicService trackPositionSortingPublicService;
    private final PlaylistPublicService playlistPublicService;
    private final BigtableTrackPositionPublicManager bigtableTrackPositionPublicManager;
    private final BigtableTrackPositionDataEnrichmentPublicService dataEnrichmentPublicService;
    private final SpotifyPlaylistLifetimePublicService spotifyPlaylistLifetimePublicService;
    private final AppleMusicPlaylistLifetimePublicService appleMusicPlaylistLifetimePublicService;

    public BigtableTrackPositionPublicService(
        BigtableTrackPositionPublicQueryExecutor trackPositionPublicQueryExecutor,
        PaginationService paginationService,
        TrackPositionConverter trackPositionConverter,
        TrackPositionPublicAggregator trackPositionPublicAggregator,
        TrackPositionSortingPublicService trackPositionSortingPublicService,
        PlaylistPublicService playlistPublicService,
        BigtableTrackPositionPublicManager bigtableTrackPositionPublicManager,
        BigtableTrackPositionDataEnrichmentPublicService dataEnrichmentPublicService,
        SpotifyPlaylistLifetimePublicService spotifyPlaylistLifetimePublicService,
        AppleMusicPlaylistLifetimePublicService appleMusicPlaylistLifetimePublicService
    ) {
        this.trackPositionPublicQueryExecutor = trackPositionPublicQueryExecutor;
        this.paginationService = paginationService;
        this.trackPositionConverter = trackPositionConverter;
        this.trackPositionPublicAggregator = trackPositionPublicAggregator;
        this.trackPositionSortingPublicService = trackPositionSortingPublicService;
        this.playlistPublicService = playlistPublicService;
        this.bigtableTrackPositionPublicManager = bigtableTrackPositionPublicManager;
        this.dataEnrichmentPublicService = dataEnrichmentPublicService;
        this.spotifyPlaylistLifetimePublicService = spotifyPlaylistLifetimePublicService;
        this.appleMusicPlaylistLifetimePublicService = appleMusicPlaylistLifetimePublicService;
    }

    @Transactional(readOnly = true)
    public CompletableFuture<ModelsListWithMetadataDto<PublicPlaylistTrackPosition, PublicPlaylistTrackPositionsMeta>>
    getPlaylistTrackPositionModels(Params params) throws NotSupportedException {

        List<String> requestedPlaylistIds = ModelUtils.filterPlaylistsByDsp(params.getPlaylistId(), params.getDsp());

        boolean noIsrcFoundForArtist = CollectionUtils.isNotEmpty(params.getArtistId()) && CollectionUtils.isEmpty(params.getIsrc());
        boolean noPlaylistIdsAfterDspFiltering = CollectionUtils.isEmpty(params.getIsrc()) && CollectionUtils.isEmpty(requestedPlaylistIds);

        if (noIsrcFoundForArtist || noPlaylistIdsAfterDspFiltering) {
            return CompletableFuture.completedFuture(
                ModelsListWithMetadataDto.<PublicPlaylistTrackPosition, PublicPlaylistTrackPositionsMeta>builder()
                    .models(Stream.empty())
                    .build());
        }

        /* If start/end date param is empty - set default */
        boolean emptyStartOrEndDate = params.getStartDate() == null || params.getEndDate() == null;

        Params paramsWithDatesSet;
        if (emptyStartOrEndDate) {
            paramsWithDatesSet = params.toBuilder()
                .startDate(DateUtils.getCurrentDate().minusDays(DAYS_SHIFT_TO_SCAN_FOR_LATEST_DATA))
                .endDate(DateUtils.getCurrentDate())
                .build();
        } else {
            paramsWithDatesSet = params;
        }

        /* Process playlistIds passed as param, so that if list contains Amazon full playlistIds (like 'amazon_123_us')
         * - they are cut to base part (like 'amazon_123'). This is needed to query BT.playlists which uses cut playlist id in rowkey
         * (query result PublicPlaylistTrackPosition objects will again have full playlistId set to them as one is expanded into multiple per-country items)
         */
        Optional<List<String>> processedPlaylistIds = ModelUtils.cutFullAmazonPlaylistIdsToBasePart(requestedPlaylistIds);

        //params with Amazon playlistIds cut from 'amazon_123_us' to 'amazon_123'
        Params paramsProcessed = paramsWithDatesSet.toBuilder()
            .playlistId(processedPlaylistIds.orElse(null))
            .build();

        EnumSet<EntityReadingParams> entityReadingParams =
            CollectionUtils.contains(paramsProcessed.getExpandTo(), ExpandTo.TRACK_POSITIONS_OF_ISRC)
                ? EnumSet.of(EntityReadingParams.EXPAND_ISRC_BY_TRACK_CURRENT_POSITION)
                : EnumSet.noneOf(EntityReadingParams.class);

        CompletableFuture<Stream<PublicPlaylistTrackPosition>> playlistTrackPositionsFuture =
            trackPositionPublicQueryExecutor.getAllPublicPlaylistTrackPositions(paramsProcessed, entityReadingParams)
                .thenApply(trackPositionConverter::convertPublicPlaylistPositions)
                .thenApply(trackPositionPublicAggregator::groupBy);

        return playlistTrackPositionsFuture
            //filter models list and populate with playlists
            .thenApply(models -> filterOnlyLatestDate(models, emptyStartOrEndDate))
            .thenApply(models -> filterByCountryCodes(models, paramsProcessed))
            .thenApply(models -> filterByAppleCountryCode(models, paramsProcessed))
            .thenApply(models -> filterByAmazonCountryCode(models, paramsProcessed))
            .thenApply(models -> filterAmazonPlaylistPositionsByRequestedFullPlaylistIds(models, requestedPlaylistIds))
            .thenApply(models -> filterPositionsByPlaylistPropertiesAndPopulatePlaylist(models, emptyStartOrEndDate, paramsProcessed))
            //populate fields that participate in sorting/paging
            .thenApply(models -> dataEnrichmentPublicService.populatePlaylistStreamsForXDaysPeriods(models, paramsProcessed))
            .thenApply(models -> dataEnrichmentPublicService.populateFieldsFromLifetimeMetrics(models, DspConstants.SPOTIFY,
                dataEnrichmentPublicService::getPlaylistTrackLifetimesSpotify,
                ModelUtils::composeSpotifyPlaylistTrackLifetimePublicIdFromPosition))
            .thenApply(models -> dataEnrichmentPublicService.populateFieldsFromLifetimeMetrics(models, DspConstants.APPLE,
                dataEnrichmentPublicService::getPlaylistTrackLifetimesAppleMusic,
                ModelUtils::composeApplePlaylistTrackLifetimePublicIdFromPosition))
            .thenApply(models -> bigtableTrackPositionPublicManager.filterDateTimeFieldByMinDate(models, paramsProcessed.getMinAddedDate(),
                PublicPlaylistTrackPosition::getLastAddedDateTime))
            .thenApply(dataEnrichmentPublicService::populateNumWeeksOnPlaylist)
            .thenApply(models -> bigtableTrackPositionPublicManager.populatePlaylistDates(models, paramsProcessed))
            //perform sorting/paging
            .thenApply(models -> trackPositionSortingPublicService.sortPlaylistPositions(models, paramsProcessed.getSortList()))
            .thenApply(models -> {
                List<PublicPlaylistTrackPosition> unpagedModels = models.collect(Collectors.toList());

                return ModelsListWithMetadataDto.<PublicPlaylistTrackPosition, PublicPlaylistTrackPositionsMeta>builder()
                    .models(paginationService.getPage(unpagedModels.stream(), paramsProcessed))
                    .meta(bigtableTrackPositionPublicManager.createPublicTrackPositionsListMeta(unpagedModels, paramsProcessed.getInclude()))
                    .build();
            })
            //populate rest of the fields
            .thenApply(modelsListDTO -> modelsListDTO.updateModelsList(
                dataEnrichmentPublicService.populateAmazonPlaylistsWithDspPlaylistId(modelsListDTO.getModels(), paramsProcessed)))
            .thenApply(modelsListDTO -> modelsListDTO.updateModelsList(
                bigtableTrackPositionPublicManager.populateNumTracks(modelsListDTO.getModels(), paramsProcessed)
            ))
            .thenApply(modelsListDTO -> modelsListDTO.updateModelsList(
                dataEnrichmentPublicService.populatePlaylistPositionTracks(modelsListDTO.getModels(), paramsProcessed)))
            .thenApply(modelsListDTO -> modelsListDTO.updateModelsList(
                dataEnrichmentPublicService.populatePlaylistPositionDspTracks(modelsListDTO.getModels(), paramsProcessed)));
    }

    @Transactional(readOnly = true)
    public List<PlaylistPublicPositionsDate> findTrackPositionsDates(String playlistId,
        Set<String> countryCodes, Pageable pageable) throws NotSupportedException {

        if (countryCodes == null) {
            countryCodes = Collections.emptySet();
        }

        String dspId;

        if (ModelUtils.isPlaylistOfDsp(playlistId, SPOTIFY)) {
            dspId = SPOTIFY;
        } else if (ModelUtils.isPlaylistOfDsp(playlistId, APPLE)) {
            dspId = APPLE;
        } else {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Provided playlist id of unsupported DSP. Supported: [spotify, apple]");
        }

        List<PlaylistPublicPositionsDate> trackPositionsDates = (SPOTIFY.equals(dspId)
            ? spotifyPlaylistLifetimePublicService.findByPlaylistIdPageable(playlistId, pageable)
            : appleMusicPlaylistLifetimePublicService.findByPlaylistIdAndCountries(playlistId, countryCodes, pageable)
        )
            .stream()
            .map(PlaylistPublicPositionsDate::from)
            .collect(Collectors.toList());

        Set<PlaylistIdCountryCodeNullableKey> positionsPlaylistKeys = trackPositionsDates.stream()
            .map(metric -> ModelUtils.composePlaylistKeyFromValuesAndDsp(
                metric.getPlaylistId(), metric.getCountryCode(), dspId
            ))
            .collect(Collectors.toSet());

        List<PublicPlaylist> playlistForDifferentCountries = playlistPublicService.findAllById(positionsPlaylistKeys);
        populatePlaylistSaveFlags(trackPositionsDates, playlistForDifferentCountries);

        Set<String> filteredCountryCodes = playlistForDifferentCountries.stream()
            .map(PublicPlaylist::getCountryCode)
            .map(JsonNullable::get)
            .collect(Collectors.toSet());

        if (SPOTIFY.equals(dspId) && !filteredCountryCodes.isEmpty()) {
            filteredCountryCodes = Set.of(CountryCodeConstant.GLOBAL);
        }

        if (!filteredCountryCodes.isEmpty()) {
            Params processParams = Params.builder()
                .startDate(LocalDate.of(2000, 1, 1))
                .endDate(DateUtils.getCurrentDate())
                .playlistId(List.of(playlistId))
                .countryCode(filteredCountryCodes)
                .dspId(dspId)
                .build();

            Map<String, List<LocalDateTime>> dspSyncDateTimeByCountryCode =
                trackPositionPublicQueryExecutor.getPlaylistDailyStats(processParams).join()
                    .collect(Collectors.groupingBy(PlaylistPublicDaily::getCountryCode,
                        Collectors.mapping(PlaylistPublicDaily::getDspSyncDateTime, Collectors.toList())));

            trackPositionsDates.forEach(p -> {
                String countryCode = p.getDsp().equals(SPOTIFY) ? CountryCodeConstant.GLOBAL : p.getCountryCode();
                if (dspSyncDateTimeByCountryCode.containsKey(countryCode)) {
                    List<LocalDateTime> dspSyncDateTime = dspSyncDateTimeByCountryCode.get(countryCode).stream()
                        .filter(Objects::nonNull)
                        .sorted(Comparator.reverseOrder())
                        .collect(Collectors.toList());
                    p.setDspSyncDateTime(dspSyncDateTime);
                }
            });
        }

        return trackPositionsDates;
    }

    private void populatePlaylistSaveFlags(List<PlaylistPublicPositionsDate> trackPositionsDates, List<PublicPlaylist> allPlaylists) {
        Map<PlaylistIdCountryCodeNullableKey, PublicPlaylist> playlistsById = allPlaylists
            .stream()
            .collect(Collectors.toMap(
                ModelUtils::composePlaylistKeyFromPlaylist,
                Function.identity(),
                (p1, p2) -> p1)
            );

        trackPositionsDates.forEach(positionDate -> {
            PublicPlaylist playlist = playlistsById.get(ModelUtils.composePlaylistKeyFromModel(positionDate));

            PlaylistSaveFlags playlistSaveFlags = PlaylistSaveFlags.builder()
                .saveHistory(playlist.isSaveHistory())
                .saveTrackList(playlist.isSaveTrackList())
                .build();
            positionDate.setPlaylist(playlistSaveFlags);
        });
    }

    private Stream<PublicPlaylistTrackPosition> filterOnlyLatestDate(Stream<PublicPlaylistTrackPosition> models,
        boolean emptyStartOrEndDate) {
        if (!emptyStartOrEndDate) {
            return models;
        }
        List<PublicPlaylistTrackPosition> modelsList = models.collect(Collectors.toList());

        if (CollectionUtils.isEmpty(modelsList)) {
            return Stream.empty();
        }

        return ModelUtils.filterForLatestDate(modelsList.stream(), PublicPlaylistTrackPosition::getDsp, PublicPlaylistTrackPosition::getDate);
    }

    /**
     * If any Amazon full playlistIds (like 'amazon_123_us') were passed in params - method ensures that we return only positions with exactly
     * that playlistIds (and not with same base playlist id part but different country) If any Amazon base playlistIds (like 'amazon_123') were
     * passed in params - all playlists found by that base playlist id will be returned
     */
    private Stream<PublicPlaylistTrackPosition> filterAmazonPlaylistPositionsByRequestedFullPlaylistIds(
        Stream<PublicPlaylistTrackPosition> models,
        List<String> requestedPlaylistIds
    ) {
        if (CollectionUtils.isEmpty(requestedPlaylistIds)) {
            return models;
        } else {
            return models.filter(position -> {
                if (DspConstants.AMAZON.equals(position.getDsp())
                    && ModelUtils.isAmazonFullPlaylistId(position.getPlaylistId())) {

                    //return true if:
                    // - full playlistId from this position is among requested playlistIds
                    // - base playlistId from this position (extracted from full playlistId) is among requested playlistIds
                    return requestedPlaylistIds.contains(position.getPlaylistId())
                        || requestedPlaylistIds.contains(position.getBasePlaylistIdFromAmazonFullPlaylistId());
                }

                return true;
            });
        }
    }

    private Stream<PublicPlaylistTrackPosition> filterByCountryCodes(
        Stream<PublicPlaylistTrackPosition> models, Params params
    ) {
        if (CollectionUtils.isNotEmpty(params.getCountryCode())) {
            return models.filter(p -> params.getCountryCode().contains(p.getCountryCode()));
        } else {
            return models;
        }
    }

    private Stream<PublicPlaylistTrackPosition> filterByAppleCountryCode(
        Stream<PublicPlaylistTrackPosition> models, Params params
    ) {
        if (CollectionUtils.isNotEmpty(params.getAppleCountryCode())) {
            return models.filter(p -> !DspConstants.APPLE.equals(p.getDsp())
                || params.getAppleCountryCode().contains(p.getCountryCode()));
        } else {
            return models;
        }
    }

    private Stream<PublicPlaylistTrackPosition> filterByAmazonCountryCode(
        Stream<PublicPlaylistTrackPosition> models, Params params
    ) {
        if (CollectionUtils.isNotEmpty(params.getAmazonCountryCode())) {
            return models.filter(p -> !DspConstants.AMAZON.equals(p.getDsp())
                || params.getAmazonCountryCode().contains(p.getCountryCode()));
        } else {
            return models;
        }
    }

    private Stream<PublicPlaylistTrackPosition> filterPositionsByPlaylistPropertiesAndPopulatePlaylist(
        Stream<PublicPlaylistTrackPosition> modelsStream, boolean emptyStartOrEndDate, Params params
    ) {
        List<PublicPlaylistTrackPosition> models = modelsStream.collect(Collectors.toList());

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

        LocalDate latestAvailableDate;

        if (!emptyStartOrEndDate) {
            latestAvailableDate = DateUtils.getCurrentDate();
        } else {
            latestAvailableDate = models.stream()
                .map(PublicPlaylistTrackPosition::getDate)
                .max(Comparator.naturalOrder())
                .orElseGet(LocalDate::now);
        }

        return bigtableTrackPositionPublicManager.populatePlaylistsFiltered(
                models.stream(),
                params.getInclude(),
                PublicPlaylistLookupFilters.builder()
                    .requireSaveHistory(false) //allow false for some positions, filtered later
                    .requireSaveTrackList(true)
                    .includeIgnored(params.getIncludeIgnored())
                    .includeRemoved(params.getIncludeRemoved())
                    .playlistTypes(params.getPlaylistTypes())
                    .playlistOwnerCategoryId(params.getPlaylistOwnerCategoryId())
                    .playlistNameContains(params.getPlaylistNameContains())
                    .build()
            )
            //ignore position which playlist haven't passed filters
            .filter(position -> position.getTransientPlaylist() != null)
            //ignore position which playlist has saveHistory=false, UNLESS this position is at 'latestAvailableDate`
            .filter(position -> position.getTransientPlaylist().isSaveHistory()
                || latestAvailableDate.equals(position.getDate()));
    }
}