package io.delphiplatform.api.v3.bigtable;

import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

import javax.transaction.NotSupportedException;

import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.v3.bigtable.config.RequestRowKeyHandle;
import io.delphiplatform.api.v3.bigtable.config.RowKeyAggPeriod;
import io.delphiplatform.api.v3.bigtable.config.RowKeyGroup;
import io.delphiplatform.api.v3.bigtable.config.RowKeyRequestType;
import io.delphiplatform.api.v3.bigtable.entity.TikTokTopIsrcSoundsEntity;
import io.delphiplatform.api.v3.bigtable.processing.PaginationService;
import io.delphiplatform.api.v3.bigtable.processing.TikTokIsrcTopSoundAggregator;
import io.delphiplatform.api.v3.bigtable.processing.TikTokTopIsrcSoundsConverter;
import io.delphiplatform.api.v3.bigtable.processing.TikTokTopIsrcSoundsSortingService;
import io.delphiplatform.api.v3.constant.BigtableTableName;
import io.delphiplatform.api.v3.model.tiktok.top.sound.TikTokTopIsrcSound;
import io.delphiplatform.api.v3.model.tiktok.top.sound.TikTokTopIsrcSoundsItems;
import io.delphiplatform.api.v3.model.video.ExpandTo;
import io.delphiplatform.api.v3.rdb.service.TrackService;
import io.delphiplatform.api.v3.view.util.Params;

@Service
public class BigtableTikTokIsrcTopSoundService {


    private final BigtableQueryExecutor queryExecutor;
    private final BigtableRowReader rowReader;
    private final TikTokTopIsrcSoundsConverter converter;
    private final TikTokIsrcTopSoundAggregator aggregator;
    private final PaginationService paginationService;

    private final TikTokTopIsrcSoundsSortingService sortingService;
    private final TrackService trackService;

    public BigtableTikTokIsrcTopSoundService(BigtableQueryExecutor queryExecutor, BigtableRowReader rowReader,
        TikTokTopIsrcSoundsConverter converter, PaginationService paginationService, TikTokTopIsrcSoundsSortingService sortingService,
        TrackService trackService, TikTokIsrcTopSoundAggregator aggregator) {
        this.queryExecutor = queryExecutor;
        this.rowReader = rowReader;
        this.converter = converter;
        this.paginationService = paginationService;
        this.sortingService = sortingService;
        this.trackService = trackService;
        this.aggregator = aggregator;
    }

    public CompletableFuture<TikTokTopIsrcSoundsItems> getTopSoundsForIsrc(Params params) {

        List<RequestRowKeyHandle> rowKeys;

        try {
            List<String> isrcsToRequest =
                CollectionUtils.contains(params.getExpandTo(), ExpandTo.RELATED_ISRCS) ? getRelated(params.getIsrc()) : params.getIsrc();
            Params paramsToGenerateRowKeys = params.toBuilder().isrc(isrcsToRequest).build();
            RowKeyGenerator rowKeyGenerator = new RowKeyGenerator(paramsToGenerateRowKeys, RowKeyGroup.TIKTOK_TOP_ISRC_TRACKS,
                RowKeyAggPeriod.WEEK, RowKeyRequestType.RANGE);
            rowKeys = rowKeyGenerator.getRequestRowKeyHandles();
        } catch (NotSupportedException e) {
            throw new RuntimeException(String.format("Cannot generate ranges for params=%s", params), e);
        }

        return queryExecutor.executePartitioned(BigtableTableName.TIKTOK_TOP_ISRC_SOUNDS, rowKeys)
            .thenApply(rowStream -> rowReader.readRows(TikTokTopIsrcSoundsEntity.class, rowStream))
            .thenApply(converter::convert)
            .thenApply(aggregator::groupBy)
            .thenApply(sortingService::sort).thenApply(models -> paginationService.getPage(models, params))
            .thenApply(models -> models.collect(Collectors.toList())).thenApply(this::populatePositions)
            .thenApply(TikTokTopIsrcSoundsItems::of);
    }

    private List<String> getRelated(List<String> isrcs) {
        List<String> related = List.copyOf(trackService.findRelatedIsrc(isrcs));
        if (CollectionUtils.isEmpty(related)) {
            return isrcs;
        }
        return related;
    }

    private List<TikTokTopIsrcSound> populatePositions(List<TikTokTopIsrcSound> models) {
        for (int i = 0; i < models.size(); i++) {
            TikTokTopIsrcSound items = models.get(i);
            items.setPosition(i);
        }
        return models;
    }
}