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

import com.google.cloud.bigtable.data.v2.models.Row;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.sonymusic.delphi.etl.apps.proto.DapdPlaylists.DapdPlaylistTrackCountryStats;
import com.sonymusic.delphi.etl.apps.proto.DapdPlaylists.DapdPlaylistTrackPositionsCountryStats;
import com.sonymusic.delphi.etl.apps.proto.DapdPlaylists.DapdPlaylistTrackPositionsStats;
import com.sonymusic.delphi.etl.apps.proto.DapdPlaylists.DapdPlaylistTrackStats;

import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import io.delphiplatform.api.util.CollectionUtils;
import io.delphiplatform.api.util.DateUtils;
import io.delphiplatform.api.util.model.ModelUtils;
import io.delphiplatform.api.v3.bigtable.ParseProtoException;
import io.delphiplatform.api.v3.bigtable.entity.PlaylistPublicPosition;
import io.delphiplatform.api.v3.constant.DspConstants;
import lombok.extern.slf4j.Slf4j;

import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.DSP;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.ISRC;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PLAYLIST_ID;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PLAYLIST_ISRC_TRACK_POSITIONS;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PLAYLIST_TRACK;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableEntityReader.EntityReadingParams.EXPAND_ISRC_BY_TRACK_CURRENT_POSITION;

@Slf4j
@Service
public class PlaylistPublicPositionReader extends BigtableEntityReader<PlaylistPublicPosition> {

    private static final int LAST_14_DAYS_DATE_OFFSET = 13;

    @Override
    public PlaylistPublicPosition readRow(Row row) {
        PlaylistPublicPosition playlistPublicPosition = new PlaylistPublicPosition();

        playlistPublicPosition.setDate(readDate(row));
        playlistPublicPosition.setDsp(readString(row, DSP));
        playlistPublicPosition.setPlaylistId(readString(row, PLAYLIST_ID));
        playlistPublicPosition.setIsrc(readString(row, ISRC));

        readBytes(row, PLAYLIST_TRACK).ifPresent(
            bytes -> playlistPublicPosition.setDapdPlaylistTrackCountryStats(parsePlaylistTrackCountryStats(bytes))
        );

        readBytes(row, PLAYLIST_ISRC_TRACK_POSITIONS).ifPresent(
            bytes -> playlistPublicPosition.setDapdPlaylistTrackPositionsCountryStats(parsePlaylistIsrcTracksStats(bytes))
        );

        return playlistPublicPosition;
    }

    /**
     * Converts position rows to PlaylistPublicPosition.
     * <p>
     * Special handling for DSP with per-country playlist stats available in protobuf field (for 'public data' - Spotify, Apple, Amazon) - one
     * position row is expanded into many PlaylistPublicPosition's since one row will contain protobuf map with stats for different country
     * variants of this playlist
     * <p>
     * And there is 2nd level of expansion for rows that has not only per-country playlist stats map but also a per-current-position stats map
     * Then per-country PlaylistPublicPosition is further expanded into multiple per-country-and-position PlaylistPublicPosition's
     */
    @Override
    public Stream<PlaylistPublicPosition> readRows(Stream<Row> rowsStream, EnumSet<EntityReadingParams> entityReadingParams) {
        List<PlaylistPublicPosition> rawPositions = rowsStream.map(this::readRow).collect(Collectors.toList());
        List<PlaylistPublicPosition> expandedPositions = new ArrayList<>();

        rawPositions.forEach(position -> {
            Map<String, DapdPlaylistTrackStats> playlistTrackCountryStatsMap = position.getPlaylistTrackCountryStatsMap();
            Map<String, DapdPlaylistTrackPositionsStats> playlistIsrcPositionsStatsMap = position.getPlaylistIsrcTracksStatsMap();

            if (CollectionUtils.isEmpty(playlistTrackCountryStatsMap)) {
                expandedPositions.add(position);
            } else {
                /* If position's DSP has per-country playlist stats available in protobuf field (for 'public data' - must be available for Spotify, Apple, Amazon)
                   expand position into multiple */

                playlistTrackCountryStatsMap.forEach((playlistCountryCode, countryPlaylistData) -> {
                    if (Boolean.TRUE.equals(
                        ModelUtils.extractBoolean(countryPlaylistData.getIsDeleted())
                    )) {
                        return;
                    }

                    PlaylistPublicPosition perCountryCopy = new PlaylistPublicPosition(position);

                    if (DspConstants.AMAZON.equals(position.getDsp())) {
                        //for Amazon - Chartmetric playlistId is appended with countryCode like 'amazon_123' + '_us'
                        perCountryCopy.setPlaylistId(
                            ModelUtils.getPlaylistIdWithCountryCode(position.getPlaylistId(), playlistCountryCode)
                        );
                    }

                    perCountryCopy.setCountryCode(playlistCountryCode);

                    perCountryCopy.setCurrent(ModelUtils.extractLong(countryPlaylistData.getCurrentPosition()));
                    perCountryCopy.setPrevious(ModelUtils.extractLong(countryPlaylistData.getPreviousPosition()));
                    perCountryCopy.setEarliestPosition(ModelUtils.extractLong(countryPlaylistData.getEarliestPosition()));

                    /* Some of PlaylistPublicPosition fields set below may be always null in BT protobuf map, depending on DSP (Amazon has more fields set)
                     * so for some - there may be logic to calculate values in place. And some may be even set (overridden)
                     * for this PlaylistPublicPosition in external code after object is returned from this method
                     *  */

                    perCountryCopy.setTrend(ModelUtils.extractLong(countryPlaylistData.getTrend()));
                    //if trend value is empty in protobuf (but current/previous are not) - calculate in place
                    if (perCountryCopy.getTrend() == null) {
                        perCountryCopy.setTrend(ModelUtils.calculatePlaylistTrackPositionTrend(
                            perCountryCopy.getPrevious(), perCountryCopy.getCurrent()));
                    }

                    perCountryCopy.setDspTrackId(ModelUtils.extractString(countryPlaylistData.getTrackId()));
                    //daysInPlaylist may later be overridden using lifetime metrics for Spotify/Apple
                    perCountryCopy.setDaysInPlaylist(ModelUtils.extractLong(countryPlaylistData.getDaysInPlaylist()));

                    LocalDate previousPositionDate = DateUtils.parseSafe(ModelUtils.extractString(countryPlaylistData.getPreviousDate()));
                    perCountryCopy.setPreviousPositionDateTime(previousPositionDate != null ? previousPositionDate.atStartOfDay() : null);

                    perCountryCopy.setEarliestPositionDate(
                        DateUtils.parseSafe(ModelUtils.extractString(countryPlaylistData.getEarliestPositionDate())));
                    perCountryCopy.setLatestPositionDate(
                        DateUtils.parseSafe(ModelUtils.extractString(countryPlaylistData.getLatestPositionDate())));
                    perCountryCopy.setLastDateChange14Days(
                        DateUtils.parseSafe(ModelUtils.extractString(countryPlaylistData.getLastDateChange14Days())));

                    //for Amazon - take previousPositionChange14Days, trendChange14Days from BT map fields
                    if (DspConstants.AMAZON.equals(position.getDsp())) {
                        perCountryCopy.setPreviousPositionChange14Days(
                            ModelUtils.extractLong(countryPlaylistData.getPreviousPositionChange14Days()));
                        perCountryCopy.setTrendChange14Days(ModelUtils.extractLong(countryPlaylistData.getTrendChange14Days()));
                    } else {
                        //for Spotify/Apple we don't have values available in BT map, so we calculate them:

                        // if previousPositionDate is older than 14 days just set previousPositionChange14Days, trendChange14Days to null,
                        if (previousPositionDate == null
                            || previousPositionDate.isBefore(DateUtils.getCurrentDate().minusDays(LAST_14_DAYS_DATE_OFFSET))) {
                            perCountryCopy.setPreviousPositionChange14Days(null);
                            perCountryCopy.setTrendChange14Days(null);
                        } else {
                            // else - set field values using basic current/previous position values already retrieved above
                            perCountryCopy.setPreviousPositionChange14Days(perCountryCopy.getPrevious());
                            perCountryCopy.setTrendChange14Days(ModelUtils.calculatePlaylistTrackPositionTrend(
                                perCountryCopy.getPrevious(), perCountryCopy.getCurrent()));
                            perCountryCopy.setLastDateChange14Days(previousPositionDate);
                        }
                    }

                    if (CollectionUtils.isNotEmpty(playlistIsrcPositionsStatsMap)
                        && playlistIsrcPositionsStatsMap.containsKey(playlistCountryCode)) {
                        /* If position has both per-country stats AND per-country-and-position-number stats available in protobuf field -
                        expand even further - single 'per-country' position into multiple 'per-country-and-position-number'
                        (same trackId on multiple positions is possible at the same moment thus stats divided by position number and not by trackId)
                        e.g. for positions [0, 1, 2]:
                        date1, isrc1, playlist1, us, 0, track1
                        date1, isrc1, playlist1, us, 1, track2
                        date1, isrc1, playlist1, us, 2, track2
                        */

                        List<PlaylistPublicPosition> positionsExpandedPerCurrentPosition = new ArrayList<>();

                        playlistIsrcPositionsStatsMap.get(playlistCountryCode).getTrackPositionsMap().forEach((currentPosition, trackId) -> {
                            PlaylistPublicPosition perPositionCopy = new PlaylistPublicPosition(perCountryCopy);
                            //override only trackId, currentPosition fields for 'per-country-and-position' items.
                            // Other properties are inherited from 'per-country' item, which itself actually represents track with
                            // lowest (i.e. top) position within isrc. Thus, between all 'per-country-and-position' items
                            // of this isrc only one (that hase lowest position) will have ALL fields data adequate (corresponding to its current position)
                            // and all others will have only trackId and currentPosition fields data adequate to that exact item
                            perPositionCopy.setDspTrackId(trackId);
                            perPositionCopy.setCurrent(currentPosition);
                            perPositionCopy.setTopTrackForIsrc(false); //is later updated to 'true' for 1 of these items (top position)

                            positionsExpandedPerCurrentPosition.add(perPositionCopy);
                        });

                        //mark 'top' position within isrc
                        positionsExpandedPerCurrentPosition.stream()
                            .min(Comparator.comparing(PlaylistPublicPosition::getCurrent))
                            .ifPresent(topPosition -> topPosition.setTopTrackForIsrc(true));

                        if (CollectionUtils.contains(entityReadingParams, EXPAND_ISRC_BY_TRACK_CURRENT_POSITION)) {
                            expandedPositions.addAll(positionsExpandedPerCurrentPosition);
                        } else {
                            //if request was NOT to expand by current position - return only top position within isrc
                            expandedPositions.addAll(
                                positionsExpandedPerCurrentPosition.stream()
                                    .filter(PlaylistPublicPosition::isTopTrackForIsrc)
                                    .collect(Collectors.toList())
                            );
                        }
                    } else {
                        //if there is no per-position stats data - can't expand further and just return positions expanded per-country
                        expandedPositions.add(perCountryCopy);
                    }
                });
            }
        });

        return expandedPositions.stream();
    }

    private DapdPlaylistTrackCountryStats parsePlaylistTrackCountryStats(ByteString bytes) {
        try {
            return DapdPlaylistTrackCountryStats.parseFrom(bytes);
        } catch (InvalidProtocolBufferException e) {
            throw new ParseProtoException("Cannot parse per-country playlist stats proto.", e);
        }
    }

    private DapdPlaylistTrackPositionsCountryStats parsePlaylistIsrcTracksStats(ByteString bytes) {
        try {
            return DapdPlaylistTrackPositionsCountryStats.parseFrom(bytes);
        } catch (InvalidProtocolBufferException e) {
            throw new ParseProtoException("Cannot parse playlist isrc track positions stats proto.", e);
        }
    }

    @Override
    public Class<PlaylistPublicPosition> getParsedClass() {
        return PlaylistPublicPosition.class;
    }

}