package io.delphiplatform.api.v3.bigtable;

import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.util.DateUtils;
import io.delphiplatform.api.util.FunctionUtils;
import io.delphiplatform.api.util.logging.LoggingUtils;
import io.delphiplatform.api.v3.constant.ApplicationConstants;
import io.delphiplatform.api.v3.model.GroupByField;
import io.delphiplatform.api.v3.model.StreamModel;
import io.delphiplatform.api.v3.model.TrackStreamsInsightsItem;
import io.delphiplatform.api.v3.rdb.service.TrackStreamsInsightsAggregateService;
import io.delphiplatform.api.v3.view.util.Params;

import static io.delphiplatform.api.v3.bigtable.BigtableTrackPositionStreamsDataEnrichmentPublicService.BT_STREAMS_QUERY_EXCEPTION_MSG;
import static io.delphiplatform.api.v3.constant.DspConstants.*;
import static io.delphiplatform.api.v3.constant.DspConstants.AMAZON;
import static io.delphiplatform.api.v3.constant.DspConstants.SPOTIFY;

@Service
public class BigtableTrackStreamsInsightsService {

    private final BigtableStreamsService bigtableStreamsService;
    private final TrackStreamsInsightsAggregateService streamsInsightsAggregateService;

    public BigtableTrackStreamsInsightsService(BigtableStreamsService bigtableStreamsService,
        TrackStreamsInsightsAggregateService streamsInsightsAggregateService) {
        this.bigtableStreamsService = bigtableStreamsService;
        this.streamsInsightsAggregateService = streamsInsightsAggregateService;
    }

    public Map<String, List<TrackStreamsInsightsItem>> calculateStreamsInsights(Params params) {
        LocalDate startDate = ApplicationConstants.MIN_STREAMS_DATE;
        LocalDate endDate = DateUtils.getCurrentDate();
        List<String> dsps = CollectionUtils.isEmpty(params.getDsp()) ?
            List.of(SPOTIFY, APPLE, AMAZON) :
            params.getDsp();

        List<Params> streamsParamsForDsp = new ArrayList<>();
        dsps.forEach(
            dsp -> {
                Params streamsParams = Params.builder()
                    .startDate(startDate)
                    .endDate(endDate)
                    .isrc(params.getIsrc())
                    .dsp(Collections.singletonList(dsp))
                    .countryCode(params.getCountryCode())
                    .groupByFields(Set.of(GroupByField.DATE, GroupByField.COUNTRY))
                    .limit(Integer.MAX_VALUE)
                    .build();
                streamsParamsForDsp.add(streamsParams);
            }
        );

        List<StreamModel> models = queryStreams(streamsParamsForDsp);
        return streamsInsightsAggregateService.aggregate(models, params.getDsp());
    }

    private List<StreamModel> queryStreams(List<Params> streamsParamsForDsp) {
        if (CollectionUtils.isEmpty(streamsParamsForDsp)) {
            return Collections.emptyList();
        }
        StopWatch methodExecutionStopwatch = new StopWatch();
        methodExecutionStopwatch.start();

        List<CompletableFuture<List<StreamModel>>> futures = streamsParamsForDsp.stream()
            .map(params -> FunctionUtils.applyWrappedInRuntimeEx(bigtableStreamsService::getRows, params, BT_STREAMS_QUERY_EXCEPTION_MSG))
            .collect(Collectors.toList());

        return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
            .thenApply(ignored -> {
                LoggingUtils.stopAndLogManualPerformanceMeasurement(methodExecutionStopwatch,
                    "BigtableTrackStreamsInsightsService.queryStreams: BT query");
                return ignored;
            })
            .thenApply(ignored -> futures.stream()
                .flatMap(future -> future.join().stream()))
            .join()
            .collect(Collectors.toList());
    }
}