package io.delphiplatform.api.integration.manager;

import com.google.protobuf.ByteString;
import com.google.protobuf.StringValue;
import com.google.protobuf.UInt64Value;
import com.sonymusic.delphi.etl.apps.proto.Amazon;
import com.sonymusic.delphi.etl.apps.proto.Amazon.AmazonCountryStats;
import com.sonymusic.delphi.etl.apps.proto.Apple;
import com.sonymusic.delphi.etl.apps.proto.Apple.AppleCountryStats;
import com.sonymusic.delphi.etl.apps.proto.DapdPlaylists;
import com.sonymusic.delphi.etl.apps.proto.Playlists;
import com.sonymusic.delphi.etl.apps.proto.Spotify;
import com.sonymusic.delphi.etl.apps.proto.Spotify.SpotifyCountryStats;
import com.sonymusic.delphi.etl.apps.proto.TiktokSound.TikTokSound;
import com.sonymusic.delphi.etl.apps.proto.TiktokSound.TikTokTopSounds;

import org.apache.commons.lang3.NotImplementedException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

import io.delphiplatform.api.integration.client.TestBigtableClient;
import io.delphiplatform.api.integration.dto.CellDto;
import io.delphiplatform.api.integration.dto.PublicPlaylistPositionCountryDto;
import io.delphiplatform.api.integration.dto.RowDto;
import io.delphiplatform.api.integration.dto.StreamsCountryDto;
import io.delphiplatform.api.integration.dto.TikTokTopIsrcSoundDto;
import io.delphiplatform.api.integration.dto.TikTokTopIsrcSoundsDto;
import io.delphiplatform.api.integration.util.BigtableUtils;
import io.delphiplatform.api.utils.TestUtils;
import io.delphiplatform.api.v3.constant.BigtableTableName;
import io.delphiplatform.api.v3.constant.DspConstants;

import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.ARTIST_ID;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.BRAND;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.COUNTRIES;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.COUNTRY_CODE;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.DATE;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.DSP;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.DSP_SYNC_DATE_TIME;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.FOLLOWERS;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.ISRC;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.IS_DELETED;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.NUMBER_OF_TRACKS;
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.BigtableColumns.STREAMS;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.TOP_SOUNDS;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.TRACK_ID;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.UPC;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableEntityReader.meta;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableEntityReader.metrics;

@Component
public class TestBigtableManager {

    public static final DateTimeFormatter LOCAL_DATE_TIME_FORMAT =
        new DateTimeFormatterBuilder()
            .parseCaseInsensitive()
            .append(DateTimeFormatter.ISO_LOCAL_DATE)
            .appendLiteral(' ')
            .append(DateTimeFormatter.ISO_LOCAL_TIME)
            .toFormatter();

    @Autowired
    public TestBigtableClient bigtableClient;

    public enum TrackPositionsReqParamsSet {
        BY_ISRC, BY_PLAYLIST_ID, BY_ISRC_AND_PLAYLIST_ID
    }

    /**
     * Manually creates BT records. Avoid this method unless explicitly needed - prefer json-based BT population
     */
    public void createPlaylistPublicTrackPosition(
        TrackPositionsReqParamsSet paramsSet,
        LocalDate date,
        String dsp,
        String playlistId,
        String isrc,
        Map<String, PublicPlaylistPositionCountryDto> positionCountries,
        Map<String, Map<Long, String>> positionTrackMapPerCountry
    ) {
        //rowkey prefix depends on request params, exact value can be observed in runtime in BigtableTrackPositionPublicQueryExecutor
        String rowKey;

        switch (paramsSet) {
            case BY_ISRC:
                rowKey = String.format("dsp_isrc_date_playlist_day~%s~%s~%s", dsp, TestUtils.reverseString(isrc), date);
                break;
            case BY_PLAYLIST_ID:
                rowKey = String.format("playlist_date_isrc_day~%s~%s~%s", playlistId, date, TestUtils.reverseString(isrc));
                break;
            case BY_ISRC_AND_PLAYLIST_ID:
                throw new NotImplementedException("impl me");
            default:
                throw new IllegalArgumentException();
        }

        createPlaylistPublicTrackPosition(rowKey, date, dsp, playlistId, isrc, positionCountries, positionTrackMapPerCountry);
    }

    public void createPlaylistPublicTrackPosition(
        String rowKey,
        LocalDate date,
        String dsp,
        String playlistId,
        String isrc,
        Map<String, PublicPlaylistPositionCountryDto> positionCountries,
        Map<String, Map<Long, String>> positionTrackMapPerCountry
    ) {
        RowDto row = buildRowPlaylistPublicTrackPosition(
            date.toString(),
            dsp,
            playlistId,
            isrc,
            positionCountries,
            positionTrackMapPerCountry
        );

        bigtableClient.sendMutation(BigtableTableName.PLAYLIST_TRACKS, rowKey, row);
    }

    public void createPlaylistSpotifyDaily(
        String rowKey,
        String countryCode,
        String playlistId,
        LocalDate reportDate,
        LocalDateTime dspSyncDateTime,
        Integer numOfTracks
    ) {
        RowDto row = buildRowPlaylistDaily(countryCode,
            playlistId,
            reportDate.toString(),
            dspSyncDateTime.format(LOCAL_DATE_TIME_FORMAT),
            numOfTracks.toString());
        bigtableClient.sendMutation(BigtableTableName.PLAYLIST_SPOTIFY_DAILY, rowKey, row);
    }

    public void createPlaylistAppleDaily(
        String rowKey,
        String countryCode,
        String playlistId,
        LocalDate reportDate,
        LocalDateTime dspSyncDateTime,
        Integer numOfTracks
    ) {
        RowDto row = buildRowPlaylistDaily(countryCode,
            playlistId,
            reportDate.toString(),
            dspSyncDateTime.format(LOCAL_DATE_TIME_FORMAT),
            numOfTracks.toString());
        bigtableClient.sendMutation(BigtableTableName.PLAYLIST_APPLE_DAILY, rowKey, row);
    }

    /**
     * Manually creates BT records. Avoid this method unless explicitly needed - prefer json-based BT population
     * <p>
     * Non-public BT table
     */
    public void createPlaylistTrackPosition(
        String rowKey,
        LocalDate date,
        String dsp,
        String playlistId,
        String isrc,
        Map<String, PublicPlaylistPositionCountryDto> positionCountries
    ) {
        RowDto row = buildRowPlaylistTrackPosition(
            date.toString(),
            dsp,
            playlistId,
            isrc,
            positionCountries
        );

        bigtableClient.sendMutation(BigtableTableName.PLAYLISTS, rowKey, row);
    }

    public void createUpcBrandTagging(
        String rowKey,
        String upc,
        String brand,
        String deleted,
        String countryCodes
    ) {
        RowDto row = buildRowUpcBrandTagging(
            upc,
            brand,
            deleted,
            countryCodes
        );

        bigtableClient.sendMutation(BigtableTableName.BRAND_TAGGING_UPC, rowKey, row);
    }

    public void createTrackIdBrandTagging(
        String rowKey,
        String trackId,
        String dsp,
        String upc
    ) {
        RowDto row = buildRowTrackIdBrandTagging(
            trackId,
            dsp,
            upc
        );

        bigtableClient.sendMutation(BigtableTableName.BRAND_TAGGING_TRACK_ID, rowKey, row);
    }

    public void createSpotifyTrackStream(String rowKey,
        LocalDate date,
        String playlistId,
        String isrc,
        String trackId,
        String artistId,
        Map<String, StreamsCountryDto> streamsCountries
    ) {
        RowDto row = buildRowSpotifyTrackStream(
            date.toString(),
            playlistId,
            isrc,
            trackId,
            artistId,
            streamsCountries
        );

        bigtableClient.sendMutation(BigtableTableName.SPOTIFY_STREAMS, rowKey, row);
    }

    public void createTikTokTopIsrcSounds(TikTokTopIsrcSoundsDto sounds) {
        RowDto row = buildRowTikTokTopIsrcSounds(sounds);

        bigtableClient.sendMutation(BigtableTableName.TIKTOK_TOP_ISRC_SOUNDS, sounds.getRowKey(), row);
    }

    public void createAppleMusicTrackStream(String rowKey,
        LocalDate date,
        String playlistId,
        String isrc,
        String trackId,
        String artistId,
        Map<String, StreamsCountryDto> streamsCountries
    ) {
        RowDto row = buildRowAppleMusicTrackStream(
            date.toString(),
            playlistId,
            isrc,
            trackId,
            artistId,
            streamsCountries
        );

        bigtableClient.sendMutation(BigtableTableName.APPLE_STREAMS, rowKey, row);
    }

    public void createAmazonTrackStream(String rowKey,
        LocalDate date,
        String playlistId,
        String isrc,
        String trackId,
        String artistId,
        Map<String, StreamsCountryDto> streamsCountries
    ) {
        RowDto row = buildRowAmazonTrackStream(
            date.toString(),
            playlistId,
            isrc,
            trackId,
            artistId,
            streamsCountries
        );

        bigtableClient.sendMutation(BigtableTableName.AMAZON_STREAMS, rowKey, row);
    }

    public void createSpotifyPlaylistFollowersDaily(
        String rowKey,
        LocalDate date,
        String playlistId,
        String followers
    ) {
        RowDto row = buildRowSpotifyPlaylistFollowersDaily(
            date.toString(),
            playlistId,
            followers
        );

        bigtableClient.sendMutation(BigtableTableName.SPOTIFY_FACT_PLAYLIST_FOLLOWERS_DAILY, rowKey, row);
    }

    /**
     * Non-public playlist countries:
     * <pre>
     * {
     *     us: {
     *         trackId: 1,
     *         currentPosition: 5,
     *         previousPosition: 10,
     *         ...
     *     }
     * }
     * </pre>
     */
    private Playlists.PlaylistTrackCountryStats buildPlaylistTrackCountryStats(
        Map<String, PublicPlaylistPositionCountryDto> positionCountries
    ) {
        if (positionCountries == null) {
            return null;
        }

        Map<String, Playlists.PlaylistTrackStats> perCountryStats = new HashMap<>();

        for (String countryCode : positionCountries.keySet()) {
            PublicPlaylistPositionCountryDto positionCountry = positionCountries.get(countryCode);

            Playlists.PlaylistTrackStats.Builder countryStatsBuilder = Playlists.PlaylistTrackStats.newBuilder();

            countryStatsBuilder.setTrackId(StringValue.newBuilder().setValue(positionCountry.getTrackId()).build());
            countryStatsBuilder.setCurrentPosition(UInt64Value.newBuilder().setValue(positionCountry.getCurrentPosition()).build());
            Optional.ofNullable(positionCountry.getPreviousPosition())
                .map(prevPosition -> UInt64Value.newBuilder().setValue(prevPosition).build())
                .ifPresent(prevPosition -> countryStatsBuilder.setPreviousPosition(prevPosition));
            //... add more fields as needed

            perCountryStats.put(countryCode, countryStatsBuilder.build());
        }

        return Playlists.PlaylistTrackCountryStats.newBuilder()
            .putAllCountries(perCountryStats)
            .build();
    }

    /**
     * playlist countries:
     * <pre>
     * {
     *     us: {
     *         trackId: 1,
     *         currentPosition: 5,
     *         previousPosition: 10,
     *         ...
     *     }
     * }
     * </pre>
     */
    private DapdPlaylists.DapdPlaylistTrackCountryStats buildDapdPlaylistTrackCountryStats(
        Map<String, PublicPlaylistPositionCountryDto> positionCountries
    ) {
        if (positionCountries == null) {
            return null;
        }

        Map<String, DapdPlaylists.DapdPlaylistTrackStats> perCountryStats = new HashMap<>();

        for (String countryCode : positionCountries.keySet()) {
            PublicPlaylistPositionCountryDto positionCountry = positionCountries.get(countryCode);

            DapdPlaylists.DapdPlaylistTrackStats.Builder countryStatsBuilder = DapdPlaylists.DapdPlaylistTrackStats.newBuilder();

            countryStatsBuilder.setTrackId(StringValue.newBuilder().setValue(positionCountry.getTrackId()).build());
            countryStatsBuilder.setCurrentPosition(UInt64Value.newBuilder().setValue(positionCountry.getCurrentPosition()).build());
            Optional.ofNullable(positionCountry.getPreviousPosition())
                .map(prevPosition -> UInt64Value.newBuilder().setValue(prevPosition).build())
                .ifPresent(prevPosition -> countryStatsBuilder.setPreviousPosition(prevPosition));
            Optional.ofNullable(positionCountry.getEarliestPosition())
                .map(earliestPosition -> UInt64Value.newBuilder().setValue(earliestPosition).build())
                .ifPresent(earliestPosition -> countryStatsBuilder.setEarliestPosition(earliestPosition));
            Optional.ofNullable(positionCountry.getPreviousPositionDateTime())
                .map(ppDateTime -> StringValue.newBuilder().setValue(ppDateTime.toLocalDate().toString()).build())
                .ifPresent(ppDateTime -> countryStatsBuilder.setPreviousDate(ppDateTime));

            //... add more fields as needed

            perCountryStats.put(countryCode, countryStatsBuilder.build());
        }

        return DapdPlaylists.DapdPlaylistTrackCountryStats.newBuilder()
            .putAllCountries(perCountryStats)
            .build();
    }

    /**
     * positions/tracks by country:
     * <pre>
     * {
     *     us: {
     *         0: track1,
     *         1: track2
     *     }
     * }
     * </pre>
     */
    private DapdPlaylists.DapdPlaylistTrackPositionsCountryStats buildDapdPlaylistTrackPositionsCountryStats(
        Map<String, Map<Long, String>> positionTrackMapPerCountry
    ) {
        if (positionTrackMapPerCountry == null) {
            return null;
        }

        Map<String, DapdPlaylists.DapdPlaylistTrackPositionsStats> perCountryPositionsStats = new HashMap<>();

        for (String countryCode : positionTrackMapPerCountry.keySet()) {
            DapdPlaylists.DapdPlaylistTrackPositionsStats.Builder countryStatsBuilder = DapdPlaylists.DapdPlaylistTrackPositionsStats.newBuilder();

            countryStatsBuilder.putAllTrackPositions(
                positionTrackMapPerCountry.get(countryCode)
            );

            perCountryPositionsStats.put(countryCode, countryStatsBuilder.build());
        }

        return DapdPlaylists.DapdPlaylistTrackPositionsCountryStats.newBuilder()
            .putAllCountries(perCountryPositionsStats)
            .build();
    }

    /**
     * streams countries:
     * <pre>
     * {
     *     us: {
     *         streams: 150,
     *         ...
     *     }
     * }
     * </pre>
     */
    private SpotifyCountryStats buildSpotifyCountryStats(
        Map<String, StreamsCountryDto> streamsCountries
    ) {
        if (streamsCountries == null) {
            return null;
        }

        Map<String, Spotify.SpotifyStreamStats> perCountryStats = new HashMap<>();

        for (String countryCode : streamsCountries.keySet()) {
            StreamsCountryDto streamsCountry = streamsCountries.get(countryCode);

            Spotify.SpotifyStreamStats.Builder countryStatsBuilder = Spotify.SpotifyStreamStats.newBuilder();

            countryStatsBuilder.setStreams(streamsCountry.getStreams());

            //... add more fields as needed

            perCountryStats.put(countryCode, countryStatsBuilder.build());
        }

        return SpotifyCountryStats.newBuilder()
            .putAllCountries(perCountryStats)
            .build();
    }

    private TikTokTopSounds buildTikTokTopSounds(
        Map<Integer, TikTokTopIsrcSoundDto> topSounds
    ) {
        if (topSounds == null) {
            return null;
        }

        Map<Long, TikTokSound> topSoundsToWrite = new HashMap<>();

        for (Integer position : topSounds.keySet()) {

            TikTokTopIsrcSoundDto topSound = topSounds.get(position);

            topSoundsToWrite.put(position.longValue(), TikTokSound.newBuilder()
                    .setSoundId(topSound.getSoundId())
                    .setContentType(topSound.getContentType())
                    .setCreations(topSound.getCreations())
                    .setCreationsShare(topSound.getCreationsShare())
                .build());

        }

        return TikTokTopSounds.newBuilder()
            .putAllSounds(topSoundsToWrite)
            .build();
    }

    /**
     * streams countries:
     * <pre>
     * {
     *     us: {
     *         streams: 150,
     *         ...
     *     }
     * }
     * </pre>
     */
    private AppleCountryStats buildAppleMusicCountryStats(
        Map<String, StreamsCountryDto> streamsCountries
    ) {
        if (streamsCountries == null) {
            return null;
        }

        Map<String, Apple.AppleStreamStats> perCountryStats = new HashMap<>();

        for (String countryCode : streamsCountries.keySet()) {
            StreamsCountryDto streamsCountry = streamsCountries.get(countryCode);

            Apple.AppleStreamStats.Builder countryStatsBuilder = Apple.AppleStreamStats.newBuilder();

            countryStatsBuilder.setStreams(streamsCountry.getStreams());

            //... add more fields as needed

            perCountryStats.put(countryCode, countryStatsBuilder.build());
        }

        return AppleCountryStats.newBuilder()
            .putAllCountries(perCountryStats)
            .build();
    }

    /**
     * streams countries:
     * <pre>
     * {
     *     us: {
     *         streams: 150,
     *         ...
     *     }
     * }
     * </pre>
     */
    private AmazonCountryStats buildAmazonCountryStats(
        Map<String, StreamsCountryDto> streamsCountries
    ) {
        if (streamsCountries == null) {
            return null;
        }

        Map<String, Amazon.AmazonStreamStats> perCountryStats = new HashMap<>();

        for (String countryCode : streamsCountries.keySet()) {
            StreamsCountryDto streamsCountry = streamsCountries.get(countryCode);

            Amazon.AmazonStreamStats.Builder countryStatsBuilder = Amazon.AmazonStreamStats.newBuilder();

            countryStatsBuilder.setStreams(streamsCountry.getStreams());

            //... add more fields as needed

            perCountryStats.put(countryCode, countryStatsBuilder.build());
        }

        return AmazonCountryStats.newBuilder()
            .putAllCountries(perCountryStats)
            .build();
    }

    private RowDto buildRowPlaylistPublicTrackPosition(
        String date,
        String dsp,
        String playlistId,
        String isrc,
        Map<String, PublicPlaylistPositionCountryDto> positionCountries,
        Map<String, Map<Long, String>> positionTrackMapPerCountry
    ) {
        ByteString playlistTrackMap = BigtableUtils.toByteString(
            buildDapdPlaylistTrackCountryStats(positionCountries)
        );

        ByteString playlistIsrcTrackPositionsMap = BigtableUtils.toByteString(
            buildDapdPlaylistTrackPositionsCountryStats(positionTrackMapPerCountry)
        );

        //... add more fields as needed
        return RowDto.of(
            CellDto.of(meta, DATE, date),
            CellDto.of(meta, DSP, dsp),
            CellDto.of(meta, PLAYLIST_ID, playlistId),
            CellDto.of(meta, ISRC, isrc),
            CellDto.of(metrics, PLAYLIST_TRACK, playlistTrackMap),
            CellDto.of(metrics, PLAYLIST_ISRC_TRACK_POSITIONS, playlistIsrcTrackPositionsMap)
        );
    }

    /**
     * Non-public
     */
    private RowDto buildRowPlaylistTrackPosition(
        String date,
        String dsp,
        String playlistId,
        String isrc,
        Map<String, PublicPlaylistPositionCountryDto> positionCountries
    ) {
        ByteString playlistTrackMap = BigtableUtils.toByteString(
            buildPlaylistTrackCountryStats(positionCountries)
        );

        //... add more fields as needed
        return RowDto.of(
            CellDto.of(meta, DATE, date),
            CellDto.of(meta, DSP, dsp),
            CellDto.of(meta, PLAYLIST_ID, playlistId),
            CellDto.of(meta, ISRC, isrc),
            CellDto.of(metrics, PLAYLIST_TRACK, playlistTrackMap)
        );
    }

    private RowDto buildRowSpotifyTrackStream(
        String date,
        String playlistId,
        String isrc,
        String trackId,
        String artistId,
        Map<String, StreamsCountryDto> streamsCountries
    ) {
        ByteString streamsMap = BigtableUtils.toByteString(
            buildSpotifyCountryStats(streamsCountries)
        );

        //... add more fields as needed
        return RowDto.of(
            CellDto.of(meta, DATE, date),
            CellDto.of(meta, DSP, DspConstants.SPOTIFY),
            CellDto.of(meta, PLAYLIST_ID, playlistId),
            CellDto.of(meta, ISRC, isrc),
            CellDto.of(meta, TRACK_ID, trackId),
            CellDto.of(meta, ARTIST_ID, artistId),
            CellDto.of(metrics, STREAMS, streamsMap)
        );
    }

    private RowDto buildRowTikTokTopIsrcSounds(
        TikTokTopIsrcSoundsDto soundsDto
    ) {
        ByteString topSounds = BigtableUtils.toByteString(
            buildTikTokTopSounds(soundsDto.getSounds())
        );

        return RowDto.of(
            CellDto.of(meta, DATE, soundsDto.getDate().toString()),
            CellDto.of(meta, ISRC, soundsDto.getIsrc()),
            CellDto.of(metrics, TOP_SOUNDS, topSounds)
        );
    }

    private RowDto buildRowUpcBrandTagging(
        String upc,
        String brand,
        String deleted,
        String countryCodes
    ) {
        return RowDto.of(
            CellDto.of(metrics, UPC, upc),
            CellDto.of(metrics, BRAND, brand),
            CellDto.of(metrics, IS_DELETED, deleted),
            CellDto.of(metrics, COUNTRIES, countryCodes)
        );
    }

    private RowDto buildRowTrackIdBrandTagging(
        String trackId,
        String dsp,
        String upc
    ) {
        return RowDto.of(
            CellDto.of(metrics, TRACK_ID, trackId),
            CellDto.of(metrics, DSP, dsp),
            CellDto.of(metrics, UPC, upc)
        );
    }

    private RowDto buildRowAppleMusicTrackStream(
        String date,
        String playlistId,
        String isrc,
        String trackId,
        String artistId,
        Map<String, StreamsCountryDto> streamsCountries
    ) {
        ByteString streamsMap = BigtableUtils.toByteString(
            buildAppleMusicCountryStats(streamsCountries)
        );

        //... add more fields as needed
        return RowDto.of(
            CellDto.of(meta, DATE, date),
            CellDto.of(meta, DSP, DspConstants.APPLE),
            CellDto.of(meta, PLAYLIST_ID, playlistId),
            CellDto.of(meta, ISRC, isrc),
            CellDto.of(meta, TRACK_ID, trackId),
            CellDto.of(meta, ARTIST_ID, artistId),
            CellDto.of(metrics, STREAMS, streamsMap)
        );
    }

    private RowDto buildRowAmazonTrackStream(
        String date,
        String playlistId,
        String isrc,
        String trackId,
        String artistId,
        Map<String, StreamsCountryDto> streamsCountries
    ) {
        ByteString streamsMap = BigtableUtils.toByteString(
            buildAmazonCountryStats(streamsCountries)
        );

        //... add more fields as needed
        return RowDto.of(
            CellDto.of(meta, DATE, date),
            CellDto.of(meta, DSP, DspConstants.AMAZON),
            CellDto.of(meta, PLAYLIST_ID, playlistId),
            CellDto.of(meta, ISRC, isrc),
            CellDto.of(meta, TRACK_ID, trackId),
            CellDto.of(meta, ARTIST_ID, artistId),
            CellDto.of(metrics, STREAMS, streamsMap)
        );
    }

    private RowDto buildRowPlaylistDaily(
        String countryCode,
        String playlistId,
        String reportDate,
        String dspSyncDateTime,
        String numOfTracks
    ) {
        return RowDto.of(
            CellDto.of(meta, COUNTRY_CODE, countryCode),
            CellDto.of(meta, PLAYLIST_ID, playlistId),
            CellDto.of(meta, DATE, reportDate),
            CellDto.of(metrics, DSP_SYNC_DATE_TIME, dspSyncDateTime),
            CellDto.of(metrics, NUMBER_OF_TRACKS, numOfTracks)
        );
    }

    private RowDto buildRowSpotifyPlaylistFollowersDaily(
        String date,
        String playlistId,
        String followers
    ) {
        return RowDto.of(
            CellDto.of(meta, DATE, date),
            CellDto.of(meta, PLAYLIST_ID, playlistId),
            CellDto.of(metrics, FOLLOWERS, followers)
        );
    }
}