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

import com.sonymusic.delphi.etl.apps.proto.Amazon;
import com.sonymusic.delphi.etl.apps.proto.Amazon.AmazonAggregate.SelectionSourceType;
import com.sonymusic.delphi.etl.apps.proto.Amazon.AmazonCountryStats;
import com.sonymusic.delphi.etl.apps.proto.Amazon.AmazonStreamStats;
import com.sonymusic.delphi.etl.apps.proto.Apple.AppleCountryStats;
import com.sonymusic.delphi.etl.apps.proto.Apple.AppleMusicAggregation.AppleAgeBands;
import com.sonymusic.delphi.etl.apps.proto.Apple.AppleStreamStats;
import com.sonymusic.delphi.etl.apps.proto.Spotify.SpotifyAgeBand;
import com.sonymusic.delphi.etl.apps.proto.Spotify.SpotifyCountryStats;
import com.sonymusic.delphi.etl.apps.proto.Spotify.SpotifyStreamStats;

import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

import java.time.LocalDate;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import io.delphiplatform.api.v3.bigtable.entity.AmazonTrackStream;
import io.delphiplatform.api.v3.bigtable.entity.AppleTrackStream;
import io.delphiplatform.api.v3.bigtable.entity.SpotifyTrackStream;
import io.delphiplatform.api.v3.model.IncludeStreams;
import io.delphiplatform.api.v3.model.StreamModel;

class StreamsConverterTest {

    private final StreamsConverter converter = new StreamsConverter();

    @Test
    void convertForCountryCode_noDemographicAllCountries() {
        String testArtistId = "artist_id";
        String testCountryCode = "us";
        long testStreams = 100L;
        List<StreamModel> converted = converter.convertForCountryCode(Collections.emptySet(), Stream.of(
            createAmazonTrackStream(testArtistId, testCountryCode, testStreams)
        ), null).collect(Collectors.toList());

        Assertions.assertEquals(1, converted.size());
        StreamModel streamModel = converted.get(0);

        Assertions.assertEquals(testCountryCode, streamModel.getCountryCode());
        Assertions.assertEquals(testArtistId, streamModel.getArtistId());
        Assertions.assertEquals(testStreams, streamModel.getStreams());
    }


    @Test
    void convertForCountryCode_onlyDspStreamsInfoAllCountries() {
        String testArtistId = "artist_id";
        String testCountryCode = "us";

        long all017 = 20L;
        long unknown1824 = 30L;
        long male017 = 40L;
        long female3544 = 50L;
        long testStreams = 100L;
        long offlinePlay = 200L;

        List<StreamModel> converted = converter.convertForCountryCode(Collections.emptySet(), Stream.of(
            createAppleTrackStream(testArtistId, testCountryCode, testStreams,
                AppleAgeBands.newBuilder()
                    .setAll017(all017)
                    .setUnknown1824(unknown1824)
                    .setMale017(male017)
                    .setFemale3544(female3544)
                    .build(), offlinePlay)
        ), Collections.singleton(IncludeStreams.DSP_STREAMS_INFO)).collect(Collectors.toList());

        StreamModel streamModel = converted.get(0);

        Assertions.assertEquals(testCountryCode, streamModel.getCountryCode());
        Assertions.assertEquals(testArtistId, streamModel.getArtistId());
        Assertions.assertEquals(testStreams, streamModel.getStreams());
        Assertions.assertFalse(streamModel.getAppleAgeBands().isPresent());
        Assertions.assertEquals(offlinePlay, streamModel.getAppleStreamsInfo().get().getOfflinePlay().get());
    }

    @Test
    void convertForCountryCode_appleWithDemographics() {
        String testArtistId = "artist_id";
        String testCountryCode = "us";

        long all017 = 20L;
        long unknown1824 = 30L;
        long male017 = 40L;
        long female3544 = 50L;
        long offlinePlay = 100L;

        List<StreamModel> converted = converter.convertForCountryCode(Collections.emptySet(), Stream.of(
            createAppleTrackStream(testArtistId, testCountryCode, 0,
                AppleAgeBands.newBuilder()
                    .setAll017(all017)
                    .setUnknown1824(unknown1824)
                    .setMale017(male017)
                    .setFemale3544(female3544)
                    .build(), offlinePlay)
        ), Collections.singleton(IncludeStreams.DEMOGRAPHICS)).collect(Collectors.toList());

        Assertions.assertEquals(2, converted.size());
        StreamModel streams = converted.get(0);
        Assertions.assertEquals(0, streams.getStreams(),
            "Streams from demographics entry should be empty");
        Assertions.assertEquals(testCountryCode, streams.getCountryCode());
        Assertions.assertEquals(testArtistId, streams.getArtistId());

        StreamModel demographics = converted.get(1);

        Assertions.assertEquals(testCountryCode, demographics.getCountryCode());
        Assertions.assertEquals(testArtistId, demographics.getArtistId());
        Assertions.assertEquals(all017, demographics.getAppleAgeBands().get().getAll017());
        Assertions.assertEquals(unknown1824, demographics.getAppleAgeBands().get().getUnknown1824());
        Assertions.assertEquals(male017, demographics.getAppleAgeBands().get().getMale017());
        Assertions.assertEquals(female3544, demographics.getAppleAgeBands().get().getFemale3544());
    }

    @Test
    void convertForCountryCode_spotifyWithDemographics() {
        String testArtistId = "artist_id";
        String testCountryCode = "us";

        long all017 = 20L;
        long unknown1822 = 30L;
        long male017 = 40L;
        long female3544 = 50L;
        long offlinePlay = 100L;

        List<StreamModel> converted = converter.convertForCountryCode(Collections.emptySet(), Stream.of(
            createSpotifyTrackStream(testArtistId, testCountryCode, 0,
                SpotifyAgeBand.newBuilder()
                    .setAll017(all017)
                    .setUnknown1822(unknown1822)
                    .setMale017(male017)
                    .setFemale3544(female3544)
                    .build(), offlinePlay)
        ), Collections.singleton(IncludeStreams.DEMOGRAPHICS)).collect(Collectors.toList());

        Assertions.assertEquals(2, converted.size());
        StreamModel streams = converted.get(0);
        Assertions.assertEquals(0, streams.getStreams(),
            "Streams from demographics entry should be empty");
        Assertions.assertEquals(testCountryCode, streams.getCountryCode());
        Assertions.assertEquals(testArtistId, streams.getArtistId());

        StreamModel demographics = converted.get(1);

        Assertions.assertEquals(testCountryCode, demographics.getCountryCode());
        Assertions.assertEquals(testArtistId, demographics.getArtistId());
        Assertions.assertEquals(all017, demographics.getSpotifyAgeBands().get().getAll017());
        Assertions.assertEquals(unknown1822, demographics.getSpotifyAgeBands().get().getUnknown1822());
        Assertions.assertEquals(male017, demographics.getSpotifyAgeBands().get().getMale017());
        Assertions.assertEquals(female3544, demographics.getSpotifyAgeBands().get().getFemale3544());
    }

    @Test
    void hasStationStreams_checkFieldsSet() {
        Arrays.stream(Amazon.AmazonAggregate.SelectionSourceType.Builder.class.getDeclaredMethods())
            .filter(method -> method.getName().startsWith("set") && method.getName().toLowerCase().contains("station"))
            .forEach(method -> {
                Amazon.AmazonAggregate.SelectionSourceType.Builder builder = SelectionSourceType.newBuilder();
                try {
                    method.invoke(builder, 1);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }

                boolean methodContributedToStationStreamsDetection = converter.hasStationStreams(builder.build());

                Assertions.assertTrue(methodContributedToStationStreamsDetection,
                    String.format("Getter for '%s' is overlooked in StreamsConverter::hasStationStreams", method.getName()));
            });
    }

    private AmazonTrackStream createAmazonTrackStream(String testArtistId, String countryCode, long streams) {
        AmazonTrackStream amazonTrackStream = new AmazonTrackStream();

        amazonTrackStream.setFullDsp("amazon_music");
        amazonTrackStream.setDate(LocalDate.of(2020, 1, 1));
        amazonTrackStream.setArtistId(testArtistId);
        amazonTrackStream.setIsrc("isrc");
        amazonTrackStream.setPlaylistId("playlist_id");
        amazonTrackStream.setProductId("product_id");
        amazonTrackStream.setTrackId("track_id");
        amazonTrackStream.setStreams(AmazonCountryStats.newBuilder()
            .putCountries(countryCode, AmazonStreamStats.newBuilder()
                .setStreams(streams)
                .build())
            .build());

        return amazonTrackStream;
    }

    private AppleTrackStream createAppleTrackStream(String testArtistId, String countryCode, long streams,
        AppleAgeBands appleAgeBands, long offlinePlay) {
        AppleTrackStream appleTrackStream = new AppleTrackStream();

        appleTrackStream.setDsp("apple");
        appleTrackStream.setDate(LocalDate.of(2020, 1, 1));
        appleTrackStream.setArtistId(testArtistId);
        appleTrackStream.setIsrc("isrc");
        appleTrackStream.setPlaylistId("playlist_id");
        appleTrackStream.setProductId("product_id");
        appleTrackStream.setTrackId("track_id");

        appleTrackStream.setStreams(AppleCountryStats.newBuilder()
            .putCountries(countryCode, AppleStreamStats.newBuilder()
                .setStreams(streams)
                .setOfflinePlay(offlinePlay)
                .build())
            .build());

        appleTrackStream.setDemographics(AppleCountryStats.newBuilder()
            .putCountries(countryCode, AppleStreamStats.newBuilder()
                .setStreams(streams)
                .setAppleAgeBands(appleAgeBands)
                .build())
            .build());

        return appleTrackStream;
    }

    private SpotifyTrackStream createSpotifyTrackStream(String testArtistId, String countryCode, long streams,
        SpotifyAgeBand spotifyAgeBand, long offlinePlay) {
        SpotifyTrackStream spotifyTrackStream = new SpotifyTrackStream();

        spotifyTrackStream.setDsp("apple");
        spotifyTrackStream.setDate(LocalDate.of(2020, 1, 1));
        spotifyTrackStream.setArtistId(testArtistId);
        spotifyTrackStream.setIsrc("isrc");
        spotifyTrackStream.setPlaylistId("playlist_id");
        spotifyTrackStream.setProductId("product_id");
        spotifyTrackStream.setTrackId("track_id");

        spotifyTrackStream.setStreams(SpotifyCountryStats.newBuilder()
            .putCountries(countryCode, SpotifyStreamStats.newBuilder()
                .setStreams(streams)
                .setOfflinePlay(offlinePlay)
                .build())
            .build());

        spotifyTrackStream.setDemographics(SpotifyCountryStats.newBuilder()
            .putCountries(countryCode, SpotifyStreamStats.newBuilder()
                .setStreams(streams)
                .setAgeBand(spotifyAgeBand)
                .build())
            .build());

        return spotifyTrackStream;
    }
}