package io.delphiplatform.api.v3.bigtable;

import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.web.server.ResponseStatusException;

import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
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.v3.bigtable.processing.AmazonStationsSortingService;
import io.delphiplatform.api.v3.constant.DspConstants;
import io.delphiplatform.api.v3.model.AggBy;
import io.delphiplatform.api.v3.model.GroupByField;
import io.delphiplatform.api.v3.model.StreamModel;
import io.delphiplatform.api.v3.model.SubsetParam;
import io.delphiplatform.api.v3.model.amazon.AmazonStationsStreamsResults;
import io.delphiplatform.api.v3.model.amazon.StationStreamingFor1DaysInfo;
import io.delphiplatform.api.v3.model.amazon.StationStreamingFor7DaysInfo;
import io.delphiplatform.api.v3.model.amazon.StationsCountryStreams;
import io.delphiplatform.api.v3.model.amazon.StationsStreamsResult;
import io.delphiplatform.api.v3.model.datahealth.DataCompleteness;
import io.delphiplatform.api.v3.rdb.entity.StationEntity;
import io.delphiplatform.api.v3.rdb.repository.StationRepository;
import io.delphiplatform.api.v3.rdb.service.DataCompletenessService;
import io.delphiplatform.api.v3.view.util.Params;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class BigtableAmazonStationsStreamsService {

    private static final int DAYS_SHIFT_TO_SCAN_FOR_7_DAYS = 6;
    private static final String AMAZON_STATION_TYPE = "amazon_station";

    private final BigtableStreamsService bigtableStreamsService;
    private final DataCompletenessService dataCompletenessService;
    private final StationRepository stationRepository;
    private final AmazonStationsSortingService amazonStationsSortingService;

    public BigtableAmazonStationsStreamsService(
        BigtableStreamsService bigtableStreamsService,
        DataCompletenessService dataCompletenessService,
        StationRepository stationRepository,
        AmazonStationsSortingService amazonStationsSortingService
    ) {
        this.bigtableStreamsService = bigtableStreamsService;
        this.dataCompletenessService = dataCompletenessService;
        this.stationRepository = stationRepository;
        this.amazonStationsSortingService = amazonStationsSortingService;
    }

    public AmazonStationsStreamsResults getAmazonStationsStreams(Params params) {
        DataCompleteness lastUpdated = dataCompletenessService.findDataCompletenessByDsp(DspConstants.AMAZON);
        LocalDate latestAvailableDate = lastUpdated.getUpdatedDate().isPresent()
            ? lastUpdated.getUpdatedDate().get() : DateUtils.getCurrentDate();

        /* Load stations streams data */

        Params amazonStreamsFor1DayParams = Params.builder()
            .startDate(latestAvailableDate)
            .endDate(latestAvailableDate)
            .isrc(params.getIsrc())
            .countryCode(params.getCountryCode())
            .dsp(Collections.singletonList(DspConstants.AMAZON))
            .aggBy(AggBy.ISRC)
            .subset(SubsetParam.PLAYLISTS)
            .groupByFields(Set.of(GroupByField.DATE, GroupByField.COUNTRY))
            .limit(Integer.MAX_VALUE) //unlimited query result size
            .build();

        //find stations with >0 streams for provided isrc's at latest date we have data for
        List<StreamModel> nonEmptyStationsStreamsFor1Day = retrieveStreamsFromBT(amazonStreamsFor1DayParams)
            .stream()
            //make sure to ignore playlists AND filter only stations with >0 streams
            .filter(streamModel -> Boolean.TRUE.equals(streamModel.hasStationStreams()))
            .collect(Collectors.toList());

        if (nonEmptyStationsStreamsFor1Day.isEmpty()) {
            return new AmazonStationsStreamsResults(latestAvailableDate, Collections.emptyList());
        }

        //stations that have any streams at latest date for provided isrc's
        Set<String> selectedStationsIds = nonEmptyStationsStreamsFor1Day.stream()
            .map(StreamModel::getPlaylistId)
            .collect(Collectors.toSet());

        //now find isrc's streams for 6 more days for selected station ids
        Params amazonStreamsFor6DaysParams = Params.builder()
            .startDate(latestAvailableDate.minusDays(DAYS_SHIFT_TO_SCAN_FOR_7_DAYS))
            .endDate(latestAvailableDate.minusDays(1)) //-1 is optimization, last day is already retrieved
            .isrc(params.getIsrc())
            .playlistId(new ArrayList<>(selectedStationsIds)) //query for isrcs and only for selected stations ids
            .countryCode(params.getCountryCode())
            .dsp(Collections.singletonList(DspConstants.AMAZON))
            .aggBy(AggBy.ISRC)
            .subset(SubsetParam.PLAYLISTS)
            .groupByFields(Set.of(GroupByField.DATE, GroupByField.COUNTRY))
            .limit(Integer.MAX_VALUE) //unlimited query result size
            .build();

        List<StreamModel> isrcStreamsFor7Days = Stream.concat(
            nonEmptyStationsStreamsFor1Day.stream(),
            retrieveStreamsFromBT(amazonStreamsFor6DaysParams).stream()
        ).collect(Collectors.toList());

        //now find streams for any isrc for 7 days from latest date at selected station ids
        Params amazonAllStreamsFor7DaysParams = Params.builder()
            .startDate(latestAvailableDate.minusDays(DAYS_SHIFT_TO_SCAN_FOR_7_DAYS))
            .endDate(latestAvailableDate)
            .playlistId(new ArrayList<>(selectedStationsIds))
            .countryCode(params.getCountryCode())
            .dsp(Collections.singletonList(DspConstants.AMAZON))
            .subset(SubsetParam.PLAYLISTS)
            .groupByFields(Set.of(GroupByField.DATE, GroupByField.COUNTRY))
            .limit(Integer.MAX_VALUE) //unlimited query result size
            .build();

        List<StreamModel> allStreamsFor7Days = retrieveStreamsFromBT(amazonAllStreamsFor7DaysParams);

        /* Transform data and prepare response */

        Map<String, Map<String, Long>> isrcStreamsSumFor1DayByStationId =
            countStreamsByStationAndCountry(nonEmptyStationsStreamsFor1Day);

        Map<String, Map<String, Long>> isrcStreamsSumFor7DaysByStationId =
            countStreamsByStationAndCountry(isrcStreamsFor7Days);

        Map<String, Map<String, Long>> allStreamsSumFor7DaysByStationId =
            countStreamsByStationAndCountry(allStreamsFor7Days);

        Map<String, Set<String>> countriesByStationId = getStreamingCountriesByStationId(
            nonEmptyStationsStreamsFor1Day);

        Map<String, StationEntity> stationMetaById = stationRepository.findAllById(selectedStationsIds)
            .stream().collect(Collectors.toMap(StationEntity::getStationId, Function.identity()));

        List<StationsStreamsResult> stationsStreams = selectedStationsIds.stream()
            .map(stationId -> {
                List<StationsCountryStreams> stationCountryStreams = new ArrayList<>();

                //prepare station's streams count for each country
                countriesByStationId.get(stationId).forEach(countryCode -> {
                    stationCountryStreams.add(
                        new StationsCountryStreams(
                            countryCode,
                            new StationStreamingFor1DaysInfo(
                                getStreamsCountForStationCountry(stationId, countryCode,
                                    isrcStreamsSumFor1DayByStationId)
                            ),
                            new StationStreamingFor7DaysInfo(
                                getStreamsCountForStationCountry(stationId, countryCode,
                                    isrcStreamsSumFor7DaysByStationId),
                                getStreamsCountForStationCountry(stationId, countryCode,
                                    allStreamsSumFor7DaysByStationId)
                            )
                        )
                    );
                });

                String stationName = stationMetaById.containsKey(stationId)
                    ? stationMetaById.get(stationId).getName() : null;

                return new StationsStreamsResult(
                    stationId,
                    stationName,
                    AMAZON_STATION_TYPE,
                    stationCountryStreams
                );
            })
            .collect(Collectors.toList());

        return new AmazonStationsStreamsResults(
            latestAvailableDate,
            amazonStationsSortingService.sortAmazonStations(
                    stationsStreams.stream(), params.getSortBy(), params.getSortOrder())
                .collect(Collectors.toList())
        );
    }

    private List<StreamModel> retrieveStreamsFromBT(Params params) {
        try {
            return bigtableStreamsService.getRows(params).get();
        } catch (NotSupportedException e) {
            throw new ResponseStatusException(HttpStatus.NOT_IMPLEMENTED, e.getMessage());
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException("Cannot get future result for BigTable query.", e);
        }
    }

    private Map<String, Map<String, Long>> countStreamsByStationAndCountry(List<StreamModel> streams) {
        Map<String, Map<String, Long>> stationIdToCountryStreams = new HashMap<>();

        streams.forEach(stream -> {
            Map<String, Long> countryToStreams = stationIdToCountryStreams.computeIfAbsent(
                stream.getPlaylistId(), playlistId -> new HashMap<>()
            );

            Long countryStreamsSum = countryToStreams.computeIfAbsent(
                stream.getCountryCode(), countryCode -> 0L
            );

            countryToStreams.put(stream.getCountryCode(), countryStreamsSum + stream.getStreams());
        });

        return stationIdToCountryStreams;
    }

    private Map<String, Set<String>> getStreamingCountriesByStationId(List<StreamModel> streams) {
        Map<String, Set<String>> stationIdToCountries = new HashMap<>();

        streams.forEach(stream -> {
            Set<String> countries = stationIdToCountries.computeIfAbsent(
                stream.getPlaylistId(), playlistId -> new HashSet<>()
            );

            countries.add(stream.getCountryCode());
        });

        return stationIdToCountries;
    }

    private Long getStreamsCountForStationCountry(
        String stationId, String countryCode, Map<String, Map<String, Long>> countryStreamsByStationId
    ) {
        if (!countryStreamsByStationId.containsKey(stationId)
            || !countryStreamsByStationId.get(stationId).containsKey(countryCode)) {
            return null;
        }

        return countryStreamsByStationId.get(stationId).get(countryCode);
    }
}