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 java.util.stream.Stream;

import javax.transaction.NotSupportedException;

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.config.RequestRowKeyHandle;
import io.delphiplatform.api.v3.bigtable.entity.Chartmetric;
import io.delphiplatform.api.v3.bigtable.processing.ArtistFollowerAggregator;
import io.delphiplatform.api.v3.bigtable.processing.ArtistFollowerConverter;
import io.delphiplatform.api.v3.bigtable.processing.ArtistFollowerSortingService;
import io.delphiplatform.api.v3.bigtable.processing.PaginationService;
import io.delphiplatform.api.v3.constant.BigtableTableName;
import io.delphiplatform.api.v3.model.ArtistFollower;
import io.delphiplatform.api.v3.view.util.Params;

@Service
public class BigtableChartmetricService {

    private final BigtableQueryExecutor queryExecutor;
    private final BigtableRowReader rowReader;
    private final ArtistFollowerConverter artistFollowerConverter;
    private final ArtistFollowerAggregator artistFollowerAggregator;
    private final ArtistFollowerSortingService artistFollowerSortingService;
    private final PaginationService paginationService;

    public BigtableChartmetricService(BigtableQueryExecutor queryExecutor, BigtableRowReader rowReader,
        ArtistFollowerConverter artistFollowerConverter,
        ArtistFollowerAggregator artistFollowerAggregator,
        ArtistFollowerSortingService artistFollowerSortingService,
        PaginationService paginationService) {
        this.queryExecutor = queryExecutor;
        this.rowReader = rowReader;
        this.artistFollowerConverter = artistFollowerConverter;
        this.artistFollowerAggregator = artistFollowerAggregator;
        this.artistFollowerSortingService = artistFollowerSortingService;
        this.paginationService = paginationService;
    }

    public CompletableFuture<List<ArtistFollower>> getModels(Params params) throws NotSupportedException {
        RowKeyGenerator rowKeyGenerator = new RowKeyGenerator(params, RowKeyGroup.CHARTMETRIC, RowKeyAggPeriod.DAILY, RowKeyRequestType.RANGE);
        List<RequestRowKeyHandle> rowKeys = rowKeyGenerator.getRequestRowKeyHandles();

        CompletableFuture<Stream<Chartmetric>> allRows = getAllChartmetric(rowKeys);

        return allRows
            .thenApply(artistFollowerConverter::convert)
            .thenApply(models -> artistFollowerAggregator.groupBy(models, params))
            .thenApply(models -> artistFollowerSortingService.sort(models, params.getSortBy(), params.getSortOrder()))
            .thenApply(models -> paginationService.getPage(models, params))
            .thenApply(models -> models.collect(Collectors.toList()));
    }

    private CompletableFuture<Stream<Chartmetric>> getAllChartmetric(List<RequestRowKeyHandle> itemKeys) {
        return queryExecutor.executePartitioned(BigtableTableName.CHARTMETRIC, itemKeys)
            .thenApply(rowStream -> rowStream.map(row -> rowReader.readRow(Chartmetric.class, row)));
    }

}