package io.delphiplatform.api.v3.bigtable;

import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.annotation.Nullable;
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.PlaylistPublicDaily;
import io.delphiplatform.api.v3.constant.CountryCodeConstant;
import io.delphiplatform.api.v3.constant.DspConstants;
import io.delphiplatform.api.v3.model.IncludeTracklistPlaylist;
import io.delphiplatform.api.v3.model.PlaylistDates;
import io.delphiplatform.api.v3.model.PlaylistDatesAwareModel;
import io.delphiplatform.api.v3.model.PublicPlaylist;
import io.delphiplatform.api.v3.model.PublicPlaylistableModel;
import io.delphiplatform.api.v3.model.trackposition.PlaylistsTrackPositionsCurrentTracklist;
import io.delphiplatform.api.v3.model.trackposition.PublicPlaylistTrackPosition;
import io.delphiplatform.api.v3.model.trackposition.PublicPlaylistTrackPositionsMeta;
import io.delphiplatform.api.v3.rdb.entity.BasePlaylistLifetimePublicEntity;
import io.delphiplatform.api.v3.rdb.entity.PlaylistDspSyncDateTimeProjection;
import io.delphiplatform.api.v3.rdb.service.apple.music.AppleMusicPlaylistLifetimePublicService;
import io.delphiplatform.api.v3.rdb.service.apple.music.AppleMusicPlaylistTrackLifetimePublicService;
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.rdb.service.spotify.SpotifyPlaylistTrackLifetimePublicService;
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 BigtableTrackPositionPublicManager {

    private static final String SPOTIFY_EXT = "spotify_ext";

    private final BigtableTrackPositionPublicQueryExecutor trackPositionPublicQueryExecutor;
    private final PlaylistPublicService playlistPublicService;

    private final SpotifyPlaylistLifetimePublicService spotifyPlaylistLifetimePublicService;
    private final AppleMusicPlaylistLifetimePublicService appleMusicPlaylistLifetimePublicService;

    private final SpotifyApiService spotifyApiService;

    private final SpotifyPlaylistTrackLifetimePublicService spotifyPlaylistTrackLifetimePublicService;
    private final AppleMusicPlaylistTrackLifetimePublicService appleMusicPlaylistTrackLifetimePublicService;

    public BigtableTrackPositionPublicManager(
        BigtableTrackPositionPublicQueryExecutor trackPositionPublicQueryExecutor,
        PlaylistPublicService playlistPublicService,
        SpotifyPlaylistLifetimePublicService spotifyPlaylistLifetimePublicService,
        AppleMusicPlaylistLifetimePublicService appleMusicPlaylistLifetimePublicService,
        SpotifyApiService spotifyApiService, SpotifyPlaylistTrackLifetimePublicService spotifyPlaylistTrackLifetimePublicService,
        AppleMusicPlaylistTrackLifetimePublicService appleMusicPlaylistTrackLifetimePublicService) {
        this.trackPositionPublicQueryExecutor = trackPositionPublicQueryExecutor;
        this.playlistPublicService = playlistPublicService;
        this.spotifyPlaylistLifetimePublicService = spotifyPlaylistLifetimePublicService;
        this.appleMusicPlaylistLifetimePublicService = appleMusicPlaylistLifetimePublicService;
        this.spotifyApiService = spotifyApiService;
        this.spotifyPlaylistTrackLifetimePublicService = spotifyPlaylistTrackLifetimePublicService;
        this.appleMusicPlaylistTrackLifetimePublicService = appleMusicPlaylistTrackLifetimePublicService;
    }

    public <P extends PublicPlaylistableModel> Stream<P> populatePlaylistsFiltered(
        Stream<P> modelsStream,
        @Nullable Collection<String> includes,
        @Nullable PublicPlaylistLookupFilters filters
    ) {
        List<P> models = modelsStream.collect(Collectors.toList());

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

        Set<PlaylistIdCountryCodeNullableKey> playlistKeys = models.stream()
            .map(ModelUtils::composePlaylistKeyFromModel)
            .collect(Collectors.toSet());

        Map<PlaylistIdCountryCodeNullableKey, PublicPlaylist> filteredPlaylistByKey =
            playlistPublicService.findAllByIdFiltered(playlistKeys, filters)
                .stream()
                .collect(Collectors.toMap(ModelUtils::composePlaylistKeyFromPlaylist, Function.identity()));

        boolean includePlaylist = CollectionUtils.contains(includes, IncludeTracklistPlaylist.PLAYLISTS.getValue());

        return models.stream()
            .peek(model -> {
                PublicPlaylist playlist = filteredPlaylistByKey.get(ModelUtils.composePlaylistKeyFromModel(model));

                if (playlist != null) {
                    model.setTransientPlaylist(playlist);

                    if (includePlaylist) {
                        model.setPlaylist(playlist);
                    }
                }
            });
    }

    public Set<String> findPlaylistIdsFilteredWithSingleAppleCountry(
        Collection<String> playlistIds,
        PublicPlaylistLookupFilters filters
    ) {
        return playlistPublicService.findPlaylistIdsFilteredWithSingleAppleCountry(playlistIds, filters);
    }

    public <P extends PlaylistDatesAwareModel> Stream<P> populatePlaylistDates(Stream<P> models, Params params) {
        if (!CollectionUtils.containsAny(params.getInclude(),
            List.of(IncludeTracklistPlaylist.PLAYLIST_DATES.getValue(), IncludeTracklistPlaylist.DSP_SYNC_DATE_TIME.getValue()))) {
            return models;
        }

        List<P> positions = models.collect(Collectors.toList());
        Map<String, List<String>> playlistIdsByDsp = positions.stream()
            .map(P::getPlaylistId)
            .filter(playlistId -> ModelUtils.isPlaylistOfDsp(playlistId, SPOTIFY) ||
                ModelUtils.isPlaylistOfDsp(playlistId, APPLE))
            .distinct()
            .collect(Collectors.groupingBy(
                playlistId -> ModelUtils.isPlaylistOfDsp(playlistId, SPOTIFY) ? SPOTIFY : APPLE));

        Map<PlaylistIdCountryCodeNullableKey, PlaylistDates> playlistDatesMap = getPlaylistDatesMap(playlistIdsByDsp, params);

        positions.forEach(position ->
            position.setPlaylistDates(playlistDatesMap
                .getOrDefault(ModelUtils.composePlaylistKeyFromValuesAndDsp(
                    position.getPlaylistId(), position.getCountryCode(), position.getDsp()
                ), null))
        );

        return positions.stream();
    }

    private Map<PlaylistIdCountryCodeNullableKey, PlaylistDates> getPlaylistDatesMap(Map<String, List<String>> playlistIdsByDsp,
        Params params) {

        Map<PlaylistIdCountryCodeNullableKey, LocalDateTime> playlistSyncDateTimeMap = getDspSyncDateTimeFromPG(playlistIdsByDsp, params);
        Map<PlaylistIdCountryCodeNullableKey, PlaylistDates> playlistDatesMap = getPlaylistLifetimeMetrics(playlistIdsByDsp, params)
            .stream()
            .collect(Collectors.toMap(entity -> ModelUtils.composePlaylistKeyFromValuesAndDsp(
                    entity.getPlaylistId(), entity.getCountryCode(), entity.getDsp()),
                entity -> PlaylistDates.from(entity,
                    playlistSyncDateTimeMap.getOrDefault(ModelUtils.composePlaylistKeyFromValuesAndDsp(
                        entity.getPlaylistId(), entity.getCountryCode(), entity.getDsp()), null)),
                (e1, e2) -> {
                    if (e1.getPlaylistUpdatedAtDateTime() == null || !e1.getPlaylistUpdatedAtDateTime().isPresent()) {
                        return e2;
                    } else if (e2.getPlaylistUpdatedAtDateTime() == null || !e2.getPlaylistUpdatedAtDateTime().isPresent()) {
                        return e1;
                    }

                    return e2.getPlaylistUpdatedAtDateTime().get().isAfter(e1.getPlaylistUpdatedAtDateTime().get()) ? e2 : e1;
                })
            );

        Map<PlaylistIdCountryCodeNullableKey, PlaylistDates> playlistSyncDatesMap = playlistSyncDateTimeMap.entrySet().stream()
            .collect(Collectors.toMap(Entry::getKey, entry -> PlaylistDates.from(entry.getValue())));

        return Stream.of(playlistDatesMap, playlistSyncDatesMap)
            .flatMap(m -> m.entrySet().stream())
            .collect(Collectors.toMap(Entry::getKey, Entry::getValue,
                (e1, e2) -> e1.getPlaylistUpdatedAtDateTime() != null || e1.getDspUpdateDateTime() != null ? e1 : e2
            ));
    }

    private Map<PlaylistIdCountryCodeNullableKey, LocalDateTime> getDspSyncDateTimeFromPG(
        Map<String, List<String>> playlistIdsByDsp, Params params) {
        if (CollectionUtils.notContains(params.getInclude(), IncludeTracklistPlaylist.DSP_SYNC_DATE_TIME.getValue())) {
            return Collections.emptyMap();
        }

        List<CompletableFuture<Stream<PlaylistDspSyncDateTimeProjection>>> futures = Collections.synchronizedList(new ArrayList<>());
        if (playlistIdsByDsp.containsKey(SPOTIFY)) {
            futures.add(CompletableFuture.supplyAsync(() ->
                spotifyPlaylistTrackLifetimePublicService.findDspSyncDateTimeForPlaylists(playlistIdsByDsp.get(SPOTIFY))
                    .stream()));
        }
        if (playlistIdsByDsp.containsKey(APPLE)) {
            futures.add(CompletableFuture.supplyAsync(() ->
                appleMusicPlaylistTrackLifetimePublicService.findDspSyncDateTimeForPlaylists(playlistIdsByDsp.get(APPLE))
                    .stream()));
        }
        return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
            .thenApply(ignored -> futures.stream().flatMap(CompletableFuture::join))
            .join()
            .collect(Collectors.toMap(item -> ModelUtils.composePlaylistKeyFromValuesAndDsp(
                    item.getPlaylistId(), item.getCountryCode(), item.getDsp()),
                PlaylistDspSyncDateTimeProjection::getDspSyncDateTime));
    }

    private List<BasePlaylistLifetimePublicEntity> getPlaylistLifetimeMetrics(
        Map<String, List<String>> playlistIdsByDsp, Params params) {
        if (CollectionUtils.notContains(params.getInclude(), IncludeTracklistPlaylist.PLAYLIST_DATES.getValue())) {
            return Collections.emptyList();
        }

        List<CompletableFuture<Stream<? extends BasePlaylistLifetimePublicEntity>>> futures =
            Collections.synchronizedList(new ArrayList<>());
        if (playlistIdsByDsp.containsKey(SPOTIFY)) {
            futures.add(CompletableFuture.supplyAsync(() ->
                spotifyPlaylistLifetimePublicService.findByPlaylistId(playlistIdsByDsp.get(SPOTIFY))
                    .stream()));
        }
        if (playlistIdsByDsp.containsKey(APPLE)) {
            futures.add(CompletableFuture.supplyAsync(() ->
                appleMusicPlaylistLifetimePublicService.findByPlaylistId(playlistIdsByDsp.get(APPLE))
                    .stream()));
        }

        return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
            .thenApply(ignored -> futures.stream().flatMap(CompletableFuture::join))
            .join()
            .collect(Collectors.toList());
    }

    public <P extends PublicPlaylistableModel> PublicPlaylistTrackPositionsMeta createPublicTrackPositionsListMeta(
        List<P> models, List<String> includes
    ) {
        if (!CollectionUtils.contains(includes, IncludeTracklistPlaylist.META_OWNER_CATEGORIES.getValue())) {
            return null;
        }

        if (models.isEmpty()) {
            return PublicPlaylistTrackPositionsMeta.builder()
                .ownerCategories(Collections.emptyList())
                .build();
        }

        return PublicPlaylistTrackPositionsMeta.builder()
            .ownerCategories(new ArrayList<>(models.stream()
                .map(position -> {
                    PublicPlaylist playlist = position.getTransientPlaylist();

                    return playlist != null && playlist.getOwner() != null && playlist.getOwner().getOwnerCategory() != null
                        ? playlist.getOwner().getOwnerCategory()
                        : null;
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toSet())))
            .build();
    }

    public <P> Stream<P> filterDateTimeFieldByMinDate(Stream<P> models,
        LocalDate minDate, Function<P, LocalDateTime> dateTimeProducer) {
        if (minDate == null) {
            return models;
        }

        return models.filter(position -> Optional.ofNullable(dateTimeProducer.apply(position))
            .map(dateTime -> dateTime.toLocalDate().isAfter(minDate.minusDays(1)))
            .orElse(Boolean.FALSE)
        );
    }

    public Stream<PublicPlaylistTrackPosition> populateNumTracks(Stream<PublicPlaylistTrackPosition> models, Params params) {
        return populateNumTracks(models, params,
            PublicPlaylistTrackPosition::getPlaylistId,
            p -> DspConstants.SPOTIFY.equals(p.getDsp())
                ? CountryCodeConstant.GLOBAL
                : p.getCountryCode(),
            PublicPlaylistTrackPosition::getDate,
            PublicPlaylistTrackPosition::setNumTracks,
            Collections.emptySet());
    }

    public Stream<PlaylistsTrackPositionsCurrentTracklist> populateNumTracks(Stream<PlaylistsTrackPositionsCurrentTracklist> models,
        Params params, Set<String> spotifyRegularPlaylists) {
        return populateNumTracks(models, params,
            PlaylistsTrackPositionsCurrentTracklist::getPlaylistId,
            p -> DspConstants.SPOTIFY.equals(p.getDsp())
                ? CountryCodeConstant.GLOBAL
                : p.getCountryCode(),
            p -> DateUtils.getCurrentDate(),
            PlaylistsTrackPositionsCurrentTracklist::setNumTracks,
            spotifyRegularPlaylists);
    }

    private <P> Stream<P> populateNumTracks(Stream<P> models, Params params,
        Function<P, String> playlistIdFactory,
        Function<P, String> countryCodeFactory,
        Function<P, LocalDate> dateFactory,
        BiConsumer<P, Long> numTrackConsumer,
        Set<String> spotifyRegularPlaylists
    ) {
        if (CollectionUtils.notContains(params.getInclude(), IncludeTracklistPlaylist.NUM_TRACKS.getValue())) {
            return models;
        }
        List<P> positions = models.collect(Collectors.toList());
        if (positions.isEmpty()) {
            return Stream.empty();
        }

        Map<String, Set<String>> playlistByDsp = positions.stream()
            .map(playlistIdFactory)
            .collect(Collectors.groupingBy(playlistId -> {
                    if (ModelUtils.isPlaylistOfDsp(playlistId, SPOTIFY)) {
                        return spotifyRegularPlaylists.contains(playlistId) ? SPOTIFY_EXT : SPOTIFY;
                    }
                    return (ModelUtils.isPlaylistOfDsp(playlistId, APPLE)) ? APPLE : "other";
                },
                Collectors.toSet()
            ));

        Function<String, String> nullCountryCodeMapper = (code) -> (code != null) ? code : "unknown";
        Set<String> countryCodes = positions.stream()
            .map(countryCodeFactory)
            .map(nullCountryCodeMapper)
            .collect(Collectors.toSet());

        TreeSet<LocalDate> orderedDate = positions.stream()
            .map(dateFactory)
            .collect(Collectors.toCollection(TreeSet::new));

        if (!(playlistByDsp.containsKey(SPOTIFY) ||
            playlistByDsp.containsKey(APPLE) ||
            !orderedDate.isEmpty() ||
            !countryCodes.isEmpty())) {
            return models;
        }

        LocalDate startDate = orderedDate.first();
        LocalDate endDate = orderedDate.last();

        List<CompletableFuture<Stream<PlaylistPublicDaily>>> futures = Collections.synchronizedList(new ArrayList<>());
        try {
            if (!playlistByDsp.getOrDefault(SPOTIFY, Collections.emptySet()).isEmpty()) {
                Params processedParams = Params.builder()
                    .startDate(startDate)
                    .endDate(endDate)
                    .countryCode(countryCodes)
                    .playlistId(new ArrayList<>(playlistByDsp.get(SPOTIFY)))
                    .dspId(SPOTIFY)
                    .build();
                futures.add(trackPositionPublicQueryExecutor.getPlaylistDailyStats(processedParams));
            }

            if (!playlistByDsp.getOrDefault(SPOTIFY_EXT, Collections.emptySet()).isEmpty()) {
                futures.add(CompletableFuture.supplyAsync(
                    () -> spotifyApiService.getPlaylistsAsPlaylistPublicDaily(playlistByDsp.get(SPOTIFY_EXT)))
                );
            }

            if (!playlistByDsp.getOrDefault(APPLE, Collections.emptySet()).isEmpty()) {
                Params processedParams = Params.builder()
                    .startDate(startDate)
                    .endDate(endDate)
                    .countryCode(countryCodes)
                    .playlistId(new ArrayList<>(playlistByDsp.get(APPLE)))
                    .dspId(APPLE)
                    .build();
                futures.add(trackPositionPublicQueryExecutor.getPlaylistDailyStats(processedParams));
            }
        } catch (NotSupportedException e) {
            throw new RuntimeException("Cannot get future result for BigTable query.", e);
        }

        Map<String, List<PlaylistPublicDaily>> playlistPublicDaily =
            CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
                .thenApply(ignored -> futures.stream().flatMap(CompletableFuture::join))
                .join()
                .collect(Collectors.groupingBy(PlaylistPublicDaily::getPlaylistId));

        positions.forEach(position -> {
            playlistPublicDaily.getOrDefault(playlistIdFactory.apply(position), Collections.emptyList())
                .stream()
                .filter(positionDaily ->
                    dateFactory.apply(position).equals(positionDaily.getReportDate()) &&
                        nullCountryCodeMapper.apply(countryCodeFactory.apply(position))
                            .equals(positionDaily.getCountryCode()))
                .findFirst()
                .ifPresent(positionDaily -> numTrackConsumer.accept(position, positionDaily.getNumberOfTracks()));
        });

        return positions.stream();
    }
}