package io.delphiplatform.api.v3.bigtable;

import com.google.common.collect.Lists;
import com.google.common.collect.Streams;

import org.apache.commons.lang3.tuple.Pair;
import org.springframework.web.server.ServerWebInputException;

import java.time.LocalDate;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import javax.transaction.NotSupportedException;

import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.util.DateUtils;
import io.delphiplatform.api.v3.bigtable.config.RowKeyAggPeriod;
import io.delphiplatform.api.v3.bigtable.config.RowKeyGroup;
import io.delphiplatform.api.v3.bigtable.config.RowKeyPrefix;
import io.delphiplatform.api.v3.bigtable.config.RowKeyRequestType;
import io.delphiplatform.api.v3.bigtable.config.RowKeySegment;
import io.delphiplatform.api.v3.bigtable.config.RequestRowKeyHandle;
import io.delphiplatform.api.v3.constant.ApplicationConstants;
import io.delphiplatform.api.v3.view.util.Params;
import lombok.NonNull;


/**
 * Creates row keys for use in Bigtable queries
 */
public class RowKeyGenerator {

    private final Params params;
    private final RowKeyGroup rowKeyGroup;
    private final RowKeyAggPeriod rowKeyAggPeriod;
    private final RowKeyPrefixHelper rowKeyPrefixHelper;
    private final RowKeyRequestType rowKeyRequestType;

    /**
     * @param params            request parameters
     * @param rowKeyGroup       group type for which to build keys
     * @param rowKeyRequestType request rowkey as exact value, as range, etc.
     * @param rowKeyAggPeriod   aggregation period
     */
    public RowKeyGenerator(@NonNull Params params, @NonNull RowKeyGroup rowKeyGroup,
        @NonNull RowKeyAggPeriod rowKeyAggPeriod, @NonNull RowKeyRequestType rowKeyRequestType) {
        this.params = params;
        this.rowKeyGroup = rowKeyGroup;
        this.rowKeyAggPeriod = rowKeyAggPeriod;
        this.rowKeyRequestType = rowKeyRequestType;
        this.rowKeyPrefixHelper = new RowKeyPrefixHelper(params, this.rowKeyGroup);
    }

    /**
     * @param params            request parameters
     * @param rowKeyGroup       group type for which to build keys
     * @param rowKeyRequestType request rowkey as exact value, as range, etc.
     */
    public RowKeyGenerator(@NonNull Params params, @NonNull RowKeyGroup rowKeyGroup, @NonNull RowKeyRequestType rowKeyRequestType) {
        this.params = params;
        this.rowKeyGroup = rowKeyGroup;
        this.rowKeyAggPeriod = RowKeyAggPeriod.DAY;
        this.rowKeyRequestType = rowKeyRequestType;
        this.rowKeyPrefixHelper = new RowKeyPrefixHelper(params, this.rowKeyGroup);
    }

    /**
     * Builds a collection of rowkey handles to query BT.
     * <p>
     * Each handle represents single "request params set" with a single rowkey or rowkey range (pair of rowkeys to e.g. get all rows between 2
     * dates). Also handle stores some additional attributes of particular query
     *
     * @return a collection of row request handles
     * @throws NotSupportedException The params object provided does not have a supported combination of fields
     */
    public List<RequestRowKeyHandle> getRequestRowKeyHandles() throws NotSupportedException {
        //if params set includes dates range - validate
        if (params.getStartDate() != null || params.getEndDate() != null) {
            checkValidDateRange();
        }

        RowKeyPrefix rowKeyPrefix = rowKeyPrefixHelper.getPrefix();
        List<String> segments = rowKeyPrefix.getSegments();

        Map<String, List<String>> startMapKeys = new LinkedHashMap<>();
        Map<String, List<String>> endMapKeys = new LinkedHashMap<>();

        for (String segment : segments) {
            List<String> paramValues = null;

            switch (segment) {
                case RowKeySegment.DATE:
                    LocalDate startDate = params.getStartDate();
                    startMapKeys.put(segment, Lists.newArrayList(DateUtils.format(startDate)));
                    // add one day to avoid trying to create end key
                    LocalDate endDate = params.getEndDate().plusDays(1);
                    endMapKeys.put(segment, Lists.newArrayList(DateUtils.format(endDate)));
                    break;
                case RowKeySegment.ISRC:
                    // Reverse ISRC segment
                    if (params.getIsrc() != null) {
                        paramValues = reverse(CollectionUtils.distinct(params.getIsrc()));
                    }
                    break;
                case RowKeySegment.VIDEO_ID:
                    // Extend to max video_id size
                    if (CollectionUtils.isNotEmpty(params.getVideoIds())) {
                        paramValues = new ArrayList<>(params.getVideoIds());
                    }
                    break;
                default:
                    paramValues = getParamAsListBySegment(segment);
            }

            if (paramValues != null) {
                startMapKeys.putIfAbsent(segment, new ArrayList<>());
                endMapKeys.putIfAbsent(segment, new ArrayList<>());

                List<String> startSegment = startMapKeys.get(segment);
                List<String> endSegment = endMapKeys.get(segment);

                startSegment.addAll(paramValues);
                endSegment.addAll(paramValues);
            }
        }

        List<Map<String, String>> startZipped = createKeyPermutations(startMapKeys);
        List<Map<String, String>> endZipped = createKeyPermutations(endMapKeys);
        if (startZipped.size() != endZipped.size()) {
            throw new AssertionError("Pre-building param lists failed: lists are different sizes");
        }

        List<Pair<String, String>> keySets = new ArrayList<>();
        String fullPrefix = rowKeyPrefix.getFullPrefix(rowKeyAggPeriod);
        for (int i = 0; i < startZipped.size(); i++) {
            List<String> startKeys = Lists.newArrayList(fullPrefix);
            List<String> endKeys = Lists.newArrayList(fullPrefix);
            for (String segment : rowKeyPrefix.getSegments()) {
                String startValue = startZipped.get(i).get(segment);
                String endValue = endZipped.get(i).get(segment);
                if (startValue != null && endValue != null) {
                    startKeys.add(startValue);
                    endKeys.add(endValue);
                }
            }
            String separator = ApplicationConstants.ROW_KEY_SEPARATOR;
            keySets.add(Pair.of(String.join(separator, startKeys), String.join(separator, endKeys)));
        }

        return keySets.stream()
            .map(pair -> RequestRowKeyHandle.builder()
                .rowKeyPrefix(rowKeyPrefix)
                .rowKey(pair)
                .rowKeyRequestType(rowKeyRequestType)
                .build())
            .collect(Collectors.toList());
    }

    private List<String> reverse(@NonNull List<String> values) {
        List<String> reversed = new ArrayList<>();
        for (String value : values) {
            reversed.add(reverse(value));
        }
        return reversed;
    }

    private String reverse(@NonNull String value) {
        return new StringBuilder(value).reverse().toString();
    }

    private List<Map<String, String>> createKeyPermutations(Map<String, List<String>> mapList) {
        List<Map<String, String>> out = new ArrayList<>();
        List<String> keys = new ArrayList<>(mapList.keySet());
        List<List<String>> products = Lists.cartesianProduct(new ArrayList<>(mapList.values()));
        for (List<String> product : products) {
            Map<String, String> zipMap = Streams.zip(keys.stream(), product.stream(), Map::of)
                .flatMap(m -> m.entrySet().stream())
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
            out.add(zipMap);
        }
        return out;
    }

    private void checkValidDateRange() {
        boolean valid = DateUtils.isValidRange(params.getStartDate(), params.getEndDate());
        if (!valid) {
            throw new ServerWebInputException("start_date is greater than end_date");
        }
    }

    private List<String> getParamAsListBySegment(String segment) throws NotSupportedException {
        List<String> result;
        switch (segment) {
            case RowKeySegment.ARTIST_ID:
                Set<String> artistIds = new HashSet<>();
                artistIds.add(params.getArtistId());

                if (params.getAdditionalArtistIds() != null) {
                    artistIds.addAll(params.getAdditionalArtistIds());
                }

                result = new ArrayList<>(artistIds);
                break;
            case RowKeySegment.PRODUCT_FAMILY_TYPE:
                result = new ArrayList<>(params.getProductFamilyIds());
                break;
            case RowKeySegment.PROJECT_NUMBER:
                result = new ArrayList<>(params.getProjectNumbers());
                break;
            case RowKeySegment.DSP:
                result = CollectionUtils.distinct(params.getDsp());
                break;
            case RowKeySegment.ISRC:
                result = CollectionUtils.distinct(params.getIsrc());
                break;
            case RowKeySegment.PLAYLIST_ID:
                result = CollectionUtils.distinct(params.getPlaylistId());
                break;
            case RowKeySegment.VIDEO_ID:
                result = new ArrayList<>(params.getVideoIds());
                break;
            case RowKeySegment.PRODUCT_ID:
                result = List.of(params.getProductId());
                break;
            case RowKeySegment.TRACK_ID:
                result = CollectionUtils.distinct(params.getTrackId());
                break;
            case RowKeySegment.UPC:
                result = CollectionUtils.distinct(params.getUpc());
                break;
            case RowKeySegment.BRAND:
                result = CollectionUtils.distinct(params.getBrand());
                break;
            case RowKeySegment.VIDEO_CONTENT_TYPE:
                result = params.getYouTubeContentTypes().stream()
                    .map(e -> e.toString().toUpperCase())
                    .sorted()
                    .collect(Collectors.toList());
                break;
            case RowKeySegment.TIKTOK_CONTENT_TYPE:
                result = params.getTikTokContentTypes().stream()
                    .map(e -> e.toString().toUpperCase())
                    .sorted()
                    .collect(Collectors.toList());
                break;
            case RowKeySegment.COUNTRY_CODE:
                result = new ArrayList<>(params.getCountryCode());
                break;
            default:
                throw new NotSupportedException("No mapping for RowKeySegment to Params getter");
        }
        return result;
    }

}