package io.delphiplatform.api.v3.bigtable;

import org.springframework.stereotype.Service;

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

import javax.transaction.NotSupportedException;

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.SpotifyPlaylistFollowersDaily;
import io.delphiplatform.api.v3.constant.BigtableTableName;
import io.delphiplatform.api.v3.view.util.Params;

@Service
public class BigtablePublicPlaylistFollowersQueryExecutor {

    private final BigtableQueryExecutor queryExecutor;
    private final BigtableRowReader rowReader;

    public BigtablePublicPlaylistFollowersQueryExecutor(BigtableQueryExecutor queryExecutor, BigtableRowReader rowReader) {
        this.queryExecutor = queryExecutor;
        this.rowReader = rowReader;
    }

    public CompletableFuture<Stream<SpotifyPlaylistFollowersDaily>> getAllSpotifyPlaylistsFollowers(Params params)
        throws NotSupportedException {
        RowKeyGenerator rowKeyGenerator = new RowKeyGenerator(params, RowKeyGroup.PLAYLIST_FOLLOWERS_DAILY, RowKeyAggPeriod.DAY,
            RowKeyRequestType.RANGE);
        List<RequestRowKeyHandle> rowKeys = rowKeyGenerator.getRequestRowKeyHandles();

        return queryExecutor.executePartitioned(BigtableTableName.SPOTIFY_FACT_PLAYLIST_FOLLOWERS_DAILY, rowKeys)
            .thenApply(rowStream -> rowReader.readRows(SpotifyPlaylistFollowersDaily.class, rowStream));
    }
}