package io.delphiplatform.api.v3.bigtable;

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

import java.time.LocalDate;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
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.DateUtils;
import io.delphiplatform.api.util.model.ModelUtils;
import io.delphiplatform.api.util.model.PlaylistIdCountryCodeNullableKey;
import io.delphiplatform.api.v3.bigtable.processing.SpotifyPlaylistFollowersConverter;
import io.delphiplatform.api.v3.constant.DspConstants;
import io.delphiplatform.api.v3.model.PublicPlaylist;
import io.delphiplatform.api.v3.model.PublicPlaylistFollowers;
import io.delphiplatform.api.v3.rdb.service.playlist_public.PlaylistPublicService;
import io.delphiplatform.api.v3.view.util.Params;

@Service
public class BigtablePublicPlaylistFollowersService {

    private static final int DAYS_SHIFT_TO_SCAN_FOR_LATEST_DATA = 30;

    private final BigtablePublicPlaylistFollowersQueryExecutor publicPlaylistFollowersQueryExecutor;
    private final SpotifyPlaylistFollowersConverter spotifyPlaylistFollowersConverter;
    private final PlaylistPublicService playlistPublicService;

    public BigtablePublicPlaylistFollowersService(
        BigtablePublicPlaylistFollowersQueryExecutor publicPlaylistFollowersQueryExecutor,
        SpotifyPlaylistFollowersConverter spotifyPlaylistFollowersConverter,
        PlaylistPublicService playlistPublicService
    ) {
        this.publicPlaylistFollowersQueryExecutor = publicPlaylistFollowersQueryExecutor;
        this.spotifyPlaylistFollowersConverter = spotifyPlaylistFollowersConverter;
        this.playlistPublicService = playlistPublicService;
    }

    @Transactional(readOnly = true)
    public CompletableFuture<List<PublicPlaylistFollowers>> getPlaylistsFollowers(Params params) throws NotSupportedException {
        LocalDate currentDate = DateUtils.getCurrentDate();

        boolean emptyStartOrEndDate = params.getStartDate() == null || params.getEndDate() == null;

        Params paramsProcessed;
        if (emptyStartOrEndDate) {
            paramsProcessed = params.toBuilder()
                .startDate(currentDate.minusDays(DAYS_SHIFT_TO_SCAN_FOR_LATEST_DATA))
                .endDate(currentDate)
                .build();
        } else {
            paramsProcessed = params.toBuilder()
                .startDate(params.getStartDate().isAfter(currentDate) ? currentDate : params.getStartDate())
                .endDate(params.getEndDate().isAfter(currentDate) ? currentDate : params.getEndDate())
                .build();
        }

        boolean isStartAtCurrentDate = currentDate.equals(paramsProcessed.getStartDate());
        boolean isEndAtCurrentDate = currentDate.equals(paramsProcessed.getEndDate());

        CompletableFuture<Stream<PublicPlaylistFollowers>> spotifyPlaylistsFollowers;

        //don't request current date from BT, value is filled from PG later on
        if (isStartAtCurrentDate && isEndAtCurrentDate) {
            spotifyPlaylistsFollowers = CompletableFuture.completedFuture(
                    spotifyPlaylistFollowersConverter.buildEmptyModels(paramsProcessed.getPlaylistId())
                )
                .thenApply(models -> fillCurrentDateFollowersWithPGData(models, currentDate));
        } else if (isEndAtCurrentDate) {
            Params paramsWithoutCurrentDate = paramsProcessed.toBuilder()
                .endDate(currentDate.minusDays(1))
                .build();

            spotifyPlaylistsFollowers =
                publicPlaylistFollowersQueryExecutor.getAllSpotifyPlaylistsFollowers(paramsWithoutCurrentDate)
                    .thenApply(spotifyPlaylistFollowersConverter::convertPlaylistFollowers)
                    //if no BT data found for some playlist at previous dates - at least create an empty model to fill data from PG for current date
                    .thenApply(models -> addEmptyModelsForCurrentDate(models, paramsProcessed.getPlaylistId()))
                    .thenApply(models -> fillCurrentDateFollowersWithPGData(models, currentDate));
        } else {
            spotifyPlaylistsFollowers =
                publicPlaylistFollowersQueryExecutor.getAllSpotifyPlaylistsFollowers(paramsProcessed)
                    .thenApply(spotifyPlaylistFollowersConverter::convertPlaylistFollowers);
        }

        return spotifyPlaylistsFollowers
            .thenApply(this::filterModelsWithEmptyFollowers)
            .thenApply(models -> models.collect(Collectors.toList()));
    }

    /**
     * If end_date is today - set followers value for today from PG instead of BT (to avoid data discrepancy)
     */
    private Stream<PublicPlaylistFollowers> fillCurrentDateFollowersWithPGData(
        Stream<PublicPlaylistFollowers> modelsStream, LocalDate currentDate
    ) {
        List<PublicPlaylistFollowers> models = modelsStream.collect(Collectors.toList());

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

        Map<PlaylistIdCountryCodeNullableKey, PublicPlaylist> playlists = playlistPublicService.findAllById(
                models.stream()
                    //once support for other DSP than Spotify is added - change playlist key generation accordingly
                    .map(model -> ModelUtils.composePlaylistKeyForSingleCountryPlaylistDsp(model.getPlaylistId(), DspConstants.SPOTIFY))
                    .collect(Collectors.toSet())
            ).stream()
            .collect(Collectors.toMap(
                ModelUtils::composePlaylistKeyFromPlaylist,
                Function.identity()
            ));

        models.forEach(model -> {
            PublicPlaylist playlist = playlists.get(
                ModelUtils.composePlaylistKeyForSingleCountryPlaylistDsp(model.getPlaylistId(), DspConstants.SPOTIFY)
            );

            Map<LocalDate, Long> followersByDate = model.getFollowersByDate();

            if (playlist != null && playlist.getFollowers() != null) {
                //replace the last list element, i.e. the latest date. Expecting that value for current date is already added to list as null
                followersByDate.put(currentDate, Long.valueOf(playlist.getFollowers()));
            }
        });

        return models.stream();
    }

    private Stream<PublicPlaylistFollowers> addEmptyModelsForCurrentDate(
        Stream<PublicPlaylistFollowers> modelsStream, Collection<String> allPlaylistIds
    ) {
        List<PublicPlaylistFollowers> models = modelsStream.collect(Collectors.toList());

        Set<String> playlistIdsFoundInBT = models.stream().map(PublicPlaylistFollowers::getPlaylistId)
            .collect(Collectors.toSet());
        Set<String> playlistIdsNotFoundInBT = new HashSet<>(allPlaylistIds);
        playlistIdsNotFoundInBT.removeAll(playlistIdsFoundInBT);

        return Stream.concat(
            models.stream(),
            spotifyPlaylistFollowersConverter.buildEmptyModels(playlistIdsNotFoundInBT)
        );
    }

    private Stream<PublicPlaylistFollowers> filterModelsWithEmptyFollowers(Stream<PublicPlaylistFollowers> models) {
        return models.filter(model -> !model.getFollowersByDate().isEmpty());
    }
}