package io.delphiplatform.api.v3.bigtable;

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

import java.time.LocalDate;
import java.util.List;
import java.util.Map;
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.processing.PaginationService;
import io.delphiplatform.api.v3.bigtable.processing.PlaylistTrackPositionAdditionPublicAggregator;
import io.delphiplatform.api.v3.bigtable.processing.PlaylistTrackPositionAdditionSortingPublicService;
import io.delphiplatform.api.v3.bigtable.processing.TrackPositionConverter;
import io.delphiplatform.api.v3.bigtable.processing.TrackPositionPublicAggregator;
import io.delphiplatform.api.v3.model.IncludeTracklistPlaylist;
import io.delphiplatform.api.v3.model.PublicPlaylist;
import io.delphiplatform.api.v3.model.trackposition.PublicPlaylistTrackPosition;
import io.delphiplatform.api.v3.model.trackposition.PublicPlaylistTrackPositionAddition;
import io.delphiplatform.api.util.model.PlaylistIdCountryCodeNullableKey;
import io.delphiplatform.api.v3.rdb.service.dto.PublicPlaylistLookupFilters;
import io.delphiplatform.api.v3.view.util.Params;

@Service
public class BigtableTrackPositionAdditionsPublicService {

    private final BigtableTrackPositionPublicQueryExecutor trackPositionPublicQueryExecutor;
    private final PaginationService paginationService;

    private final TrackPositionConverter trackPositionConverter;
    private final TrackPositionPublicAggregator trackPositionPublicAggregator;
    private final BigtableTrackPositionPublicManager bigtableTrackPositionPublicManager;
    private final PlaylistTrackPositionAdditionPublicAggregator trackPositionAdditionPublicAggregator;
    private final PlaylistTrackPositionAdditionSortingPublicService trackPositionAdditionSortingPublicService;
    private final BigtableTrackPositionDataEnrichmentPublicService dataEnrichmentPublicService;


    public BigtableTrackPositionAdditionsPublicService(
        BigtableTrackPositionPublicQueryExecutor trackPositionPublicQueryExecutor,
        PaginationService paginationService,
        TrackPositionConverter trackPositionConverter,
        TrackPositionPublicAggregator trackPositionPublicAggregator,
        BigtableTrackPositionPublicManager bigtableTrackPositionPublicManager,
        PlaylistTrackPositionAdditionPublicAggregator trackPositionAdditionPublicAggregator,
        PlaylistTrackPositionAdditionSortingPublicService trackPositionAdditionSortingPublicService,
        BigtableTrackPositionDataEnrichmentPublicService dataEnrichmentPublicService
    ) {
        this.trackPositionPublicQueryExecutor = trackPositionPublicQueryExecutor;
        this.paginationService = paginationService;
        this.trackPositionConverter = trackPositionConverter;
        this.trackPositionPublicAggregator = trackPositionPublicAggregator;
        this.bigtableTrackPositionPublicManager = bigtableTrackPositionPublicManager;
        this.trackPositionAdditionPublicAggregator = trackPositionAdditionPublicAggregator;
        this.trackPositionAdditionSortingPublicService = trackPositionAdditionSortingPublicService;
        this.dataEnrichmentPublicService = dataEnrichmentPublicService;
    }

    @Transactional(readOnly = true)
    public CompletableFuture<List<PublicPlaylistTrackPositionAddition>> getPlaylistTrackPositionAdditionModels(
        Params params)
        throws NotSupportedException {
        // use day before to determine possible position addition on start date
        Params positionParams = params.toBuilder().startDate(params.getStartDate().minusDays(1)).build();

        CompletableFuture<Stream<PublicPlaylistTrackPosition>> filteredTrackPositionsFuture =
            trackPositionPublicQueryExecutor.getAllPublicPlaylistTrackPositions(positionParams)
                .thenApply(trackPositionConverter::convertPublicPlaylistPositions)
                .thenApply(trackPositionPublicAggregator::groupBy)
                .thenApply(models -> trackPositionAdditionPublicAggregator.filterOnlyAdditions(models, positionParams));

        return filteredTrackPositionsFuture
            //filter models list and populate with playlists
            .thenApply(models -> filterPositionAdditionsByPlaylistPropertiesAndPopulatePlaylist(models, params))
            //transform positions models into additions models
            .thenApply(models -> transformTrackPositionsIntoAdditions(models, params))
            //perform sorting/paging
            .thenApply(
                models -> trackPositionAdditionSortingPublicService.sortPlaylistPositions(models, params.getSortBy(), params.getSortOrder()))
            .thenApply(models -> paginationService.getPage(models, params))
            //populate rest of the fields
            .thenApply(models -> dataEnrichmentPublicService.populateAmazonPlaylistsWithDspPlaylistId(models, params))
            .thenApply(models -> models.collect(Collectors.toList()));
    }

    private Stream<PublicPlaylistTrackPositionAddition> transformTrackPositionsIntoAdditions(
        Stream<PublicPlaylistTrackPosition> positionStream, Params params
    ) {
        List<PublicPlaylistTrackPosition> positions = positionStream.collect(Collectors.toList());

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

        //save playlists (that are already loaded for all positions)
        Map<PlaylistIdCountryCodeNullableKey, PublicPlaylist> playlistByKey = positions.stream()
            .map(PublicPlaylistTrackPosition::getTransientPlaylist)
            .distinct()
            .collect(Collectors.toMap(
                ModelUtils::composePlaylistKeyFromPlaylist,
                Function.identity()
            ));

        //transform positions into additions
        Stream<PublicPlaylistTrackPositionAddition> positionAdditions = trackPositionAdditionPublicAggregator.groupBy(positions.stream());

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

        //populate additions with playlists and return
        return positionAdditions
            .peek(model -> {
                PublicPlaylist playlist = playlistByKey.get(ModelUtils.composePlaylistKeyFromModel(model));

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

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

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

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

        LocalDate latestAvailableDate = DateUtils.getCurrentDate();

        return bigtableTrackPositionPublicManager.populatePlaylistsFiltered(
                models.stream(),
                params.getInclude(),
                PublicPlaylistLookupFilters.builder()
                    .requireSaveHistory(false) //allow false for some positions, filtered later
                    .requireSaveTrackList(true)
                    .playlistTypes(params.getPlaylistTypes())
                    .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()));
    }
}