package io.delphiplatform.api.v3.bigtable.processing;

import com.google.common.collect.Lists;

import org.springframework.stereotype.Service;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import io.delphiplatform.api.util.model.ModelUtils;
import io.delphiplatform.api.v3.bigtable.processing.groupbykey.ArtistFollowerGroupByKey;
import io.delphiplatform.api.v3.model.ArtistFollower;
import io.delphiplatform.api.v3.model.GroupByField;
import io.delphiplatform.api.v3.view.util.Params;

@Service
public class ArtistFollowerAggregator {

    public Stream<ArtistFollower> groupBy(Stream<ArtistFollower> artistFollowers, Params params) {
        List<GroupByFieldEnum> groupByFields = getGroupByFields(params);
        return groupBy(artistFollowers, groupByFields);
    }

    private Stream<ArtistFollower> groupBy(Stream<ArtistFollower> artistFollowers,
        List<GroupByFieldEnum> groupByFields) {
        return artistFollowers
            .collect(Collectors.groupingBy(followers -> new ArtistFollowerGroupByKey(followers, groupByFields)))
            .entrySet().stream()
            .map(entry -> {
                ArtistFollowerGroupByKey key = entry.getKey();
                List<ArtistFollower> group = entry.getValue();
                ArtistFollower sum = ModelUtils.mergeArtistFollowers(group);
                key.populateArtistFollowers(sum);
                return sum;
            });
    }

    private List<GroupByFieldEnum> getGroupByFields(Params params) {
        List<GroupByFieldEnum> groupByFields = Lists.newArrayList();
        if (GroupByField.DATE.equals(params.getGroupByField())) {
            groupByFields.add(GroupByFieldEnum.DATE);
        }
        return groupByFields;
    }

}