package io.delphiplatform.api.extapi;

import com.google.common.collect.Lists;

import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.stream.Stream;

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.PublicPlaylist;
import io.delphiplatform.api.v3.model.trackposition.PublicPlaylistTrackPosition;
import io.delphiplatform.api.v3.rdb.repository.spotify.SpotifyAlbumRepository;
import io.delphiplatform.api.v3.rdb.service.EntityService;
import io.delphiplatform.api.util.model.PlaylistIdCountryCodeNullableKey;
import io.delphiplatform.api.v3.rdb.service.playlist_public.PlaylistPublicService;
import lombok.extern.slf4j.Slf4j;
import se.michaelthelin.spotify.enums.ModelObjectType;
import se.michaelthelin.spotify.model_objects.specification.Album;
import se.michaelthelin.spotify.model_objects.specification.Playlist;
import se.michaelthelin.spotify.model_objects.specification.PlaylistTrack;

@Slf4j
@Service
public class SpotifyApiService {

    static final Integer SPOTIFY_ALBUM_MAX_COUNT_REQUEST = 20;

    private final PlaylistPublicService playlistPublicService;
    private final SpotifyApiPlaylistConvertor spotifyApiPlaylistConvertor;
    private final SpotifyApiPlaylistTrackConvertor spotifyApiPlaylistTrackConvertor;
    private final SpotifyApiRequestExecutor requestExecutor;
    private final SpotifyAlbumRepository spotifyAlbumRepository;

    @Autowired
    public SpotifyApiService(PlaylistPublicService playlistPublicService,
        SpotifyApiPlaylistConvertor spotifyApiPlaylistConvertor,
        SpotifyApiPlaylistTrackConvertor spotifyApiPlaylistTrackConvertor,
        SpotifyApiRequestExecutor spotifyApiRequestExecutor, SpotifyAlbumRepository spotifyAlbumRepository) {
        this.playlistPublicService = playlistPublicService;
        this.spotifyApiPlaylistConvertor = spotifyApiPlaylistConvertor;
        this.spotifyApiPlaylistTrackConvertor = spotifyApiPlaylistTrackConvertor;
        this.requestExecutor = spotifyApiRequestExecutor;
        this.spotifyAlbumRepository = spotifyAlbumRepository;
    }

    private List<Album> getAlbums(Collection<String> albumIds) {
        if (CollectionUtils.isEmpty(albumIds)) {
            return Collections.emptyList();
        }

        List<CompletableFuture<Album[]>> partitions = Lists.partition(new ArrayList<>(albumIds), SPOTIFY_ALBUM_MAX_COUNT_REQUEST)
            .stream()
            .map(c -> c.stream().toArray(String[]::new))
            .map(c -> new CompletableFuture().supplyAsync(() -> requestExecutor.getSeveralAlbums(c)))
            .collect(Collectors.toList());

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

    private Map<String, List<PlaylistTrack>> getPlaylistsItems(Collection<String> playlistIds) {
        if (CollectionUtils.isEmpty(playlistIds)) {
            return Collections.emptyMap();
        }
        List<CompletableFuture<Pair<String, List<PlaylistTrack>>>> futures = playlistIds.stream()
            .map(ModelUtils::removeDspPrefixFromPlaylistId)
            .map(playlistId -> CompletableFuture.supplyAsync(
                () -> Pair.of(playlistId, requestExecutor.getPlaylistsItems(playlistId)
                    .stream()
                    .filter(playlistTrack -> playlistTrack.getTrack() != null)
                    .filter(playlistTrack -> playlistTrack.getTrack().getType().equals(ModelObjectType.TRACK))
                    .collect(Collectors.toList())
                ))
            )
            .collect(Collectors.toList());
        return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
            .thenApply(ignored -> futures.stream().map(CompletableFuture::join))
            .join()
            .collect(Collectors.toMap(Pair::getLeft, Pair::getRight));
    }

    public Stream<PublicPlaylistTrackPosition> getPlaylistsItemsAsTrackPositions(List<String> playlistIds,
        Boolean includeTrackInfo) {
        Stream<PublicPlaylistTrackPosition> stream = getPlaylistsItems(playlistIds).entrySet().stream()
            .flatMap(entry -> spotifyApiPlaylistTrackConvertor
                .toPublicPlaylistTrackPosition(ModelUtils.addDspPrefixForPlaylistId(entry.getKey(), DspConstants.SPOTIFY), entry.getValue())
                .stream()
            );

        return includeTrackInfo
            ? spotifyApiPlaylistTrackConvertor.populateAlbumUpc(stream, this::getAlbums,
            spotifyAlbumRepository::findAlbumUpcByAlbumId)
            : stream;
    }

    public Stream<PlaylistPublicDaily> getPlaylistsAsPlaylistPublicDaily(Collection<String> playlistIds) {
        return getPlaylistsItems(playlistIds).entrySet().stream()
            .filter(entry -> !entry.getValue().isEmpty())
            .map(entry -> {
                PlaylistPublicDaily playlistPublicDaily = new PlaylistPublicDaily();
                playlistPublicDaily.setPlaylistId(ModelUtils.addDspPrefixForPlaylistId(entry.getKey(), DspConstants.SPOTIFY));
                playlistPublicDaily.setReportDate(DateUtils.getCurrentDate());
                playlistPublicDaily.setCountryCode(CountryCodeConstant.GLOBAL);
                playlistPublicDaily.setNumberOfTracks((long) entry.getValue().size());
                playlistPublicDaily.setDspSyncDateTime(DateUtils.getCurrentDateTime());
                return playlistPublicDaily;
            });
    }

    public PublicPlaylist getPlaylistAsPublicPlaylist(String playlistId) {
        String spotifyPlaylistId = ModelUtils.removeDspPrefixFromPlaylistId(playlistId);
        Playlist spotifyPlaylist = Optional.ofNullable(requestExecutor.getPlaylist(spotifyPlaylistId))
            .orElseThrow(EntityService.NOT_FOUND);

        List<PlaylistTrack> allPlaylistTracks = Arrays.asList(spotifyPlaylist.getTracks().getItems());
        if (allPlaylistTracks.size() < spotifyPlaylist.getTracks().getTotal()) {
            allPlaylistTracks = requestExecutor.getPlaylistsItems(spotifyPlaylistId);
        }

        Collection<PlaylistIdCountryCodeNullableKey> playlistKeys = Collections.singleton(
            ModelUtils.composePlaylistKeyFromValues(playlistId, null));
        PublicPlaylist publicPlaylistFromDb = playlistPublicService.findAllById(playlistKeys).stream()
            .findFirst()
            .orElse(null);
        return spotifyApiPlaylistConvertor.toPublicPlaylist(spotifyPlaylist, allPlaylistTracks, publicPlaylistFromDb);
    }
}