package spotifyconsumeranalytics

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"log"
	"math"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/filtr/go-apollo/pkg/daterange"
	pb "github.com/filtr/go-apollo/pkg/leveldbgrpc"
	"google.golang.org/grpc"
	grpcCodes "google.golang.org/grpc/codes"
)

type dbPlaylistStreamValue struct {
	PaidListeners int `json:"paid_listeners,omitempty"`
	Listeners     int `json:"listeners"`
	RepeatPlay    int `json:"repeat_play,omitempty"`
	ShufflePlay   int `json:"shuffle_play,omitempty"`
	PaidStreams   int `json:"paid_streams,omitempty"`
	Streams       int `json:"streams"`
}

type dbTrackStreamValue struct {
	PaidListeners            int `json:"paid_listeners,omitempty"`
	Listeners                int `json:"listeners"`
	RepeatPlay               int `json:"repeat_play,omitempty"`
	ShufflePlay              int `json:"shuffle_play,omitempty"`
	PaidSourceAlbum          int `json:"paid_source_album,omitempty"`
	PaidSourceArtist         int `json:"paid_source_artist,omitempty"`
	PaidSourceChart          int `json:"paid_source_chart,omitempty"`
	PaidSourceCollection     int `json:"paid_source_collection,omitempty"`
	PaidSourceDailyMix       int `json:"paid_source_daily_mix,omitempty"`
	PaidSourceDiscoverWeekly int `json:"paid_source_discover_weekly,omitempty"`
	PaidSourceOther          int `json:"paid_source_other,omitempty"`
	PaidSourceOthersPlaylist int `json:"paid_source_others_playlist,omitempty"`
	PaidSourcePlayQueue      int `json:"paid_source_play_queue,omitempty"`
	PaidSourceRadio          int `json:"paid_source_radio,omitempty"`
	PaidSourceReleaseRadar   int `json:"paid_source_release_radar,omitempty"`
	PaidSourceSearch         int `json:"paid_source_search,omitempty"`
	PaidSourceUnknown        int `json:"paid_source_unknown,omitempty"`
	SourceAlbum              int `json:"source_album,omitempty"`
	SourceArtist             int `json:"source_artist,omitempty"`
	SourceChart              int `json:"source_chart,omitempty"`
	SourceCollection         int `json:"source_collection,omitempty"`
	SourceDailyMix           int `json:"source_daily_mix,omitempty"`
	SourceDiscoverWeekly     int `json:"source_discover_weekly,omitempty"`
	SourceOther              int `json:"source_other,omitempty"`
	SourceOthersPlaylist     int `json:"source_others_playlist,omitempty"`
	SourcePlayQueue          int `json:"source_play_queue,omitempty"`
	SourceRadio              int `json:"source_radio,omitempty"`
	SourceReleaseRadar       int `json:"source_release_radar,omitempty"`
	SourceSearch             int `json:"source_search,omitempty"`
	SourceUnknown            int `json:"source_unknown,omitempty"`
	PaidStreams              int `json:"paid_streams,omitempty"`
	Streams                  int `json:"streams"`
}

type roamingTrackStream struct {
	Key   string
	Value dbTrackStreamValue
}

type dbTrackPlaylistStreamValue struct {
	PaidListeners int `json:"paid_listeners,omitempty"`
	Listeners     int `json:"listeners"`
	RepeatPlay    int `json:"repeat_play,omitempty"`
	ShufflePlay   int `json:"shuffle_play,omitempty"`
	PaidStreams   int `json:"paid_streams,omitempty"`
	Streams       int `json:"streams"`
}

type ingestValue struct {
	Timestamp string `json:"timestamp"`
}

func getRows(s3Service *s3.S3, s3TargetBucket, sourceKey string) ([]string, error) {
	input := &s3.GetObjectInput{
		Bucket: aws.String(s3TargetBucket),
		Key:    aws.String(sourceKey),
	}

	result, err := s3Service.GetObject(input)
	if err != nil {
		log.Fatal(err)
	}

	var rows []string

	scanner := bufio.NewScanner(result.Body)
	for scanner.Scan() {
		var text = scanner.Text()
		rows = append(rows, text)
	}
	if err := scanner.Err(); err != nil {
		return nil, err
	}

	return rows, nil
}

func processTrackStreamsDomestic(s3Service *s3.S3, client pb.LevelDBClient, date, licensor, reportCountryCode, s3TargetBucket string) {
	ingestKey := makeIngestTrackStreamsDomesticDBKey(date, reportCountryCode, licensor)
	if isIngested(client, ingestKey) == true {
		log.Println("track_streams_domestic - already ingested", date, licensor, reportCountryCode)
		return
	}

	sourceKey := makeSpotifyTargetStreamsTracksS3Key(date, licensor, reportCountryCode)
	if ok, err := targetFileExists(s3Service, s3TargetBucket, sourceKey); err != nil {
		log.Printf("processTrackStreamsDomestic, error: %s", err)
		return
	} else if !ok {
		log.Println("processTrackStreamsDomestic, track_streams_domestic - source doesn't exist", date, licensor, reportCountryCode, sourceKey)
		return
	}

	log.Println("track_streams_domestic - starting", date, licensor, reportCountryCode, sourceKey)

	rows, err := getRows(s3Service, s3TargetBucket, sourceKey)
	if err != nil {
		log.Fatal(err)
	}

	var list []*pb.BatchPutItem

	for _, row := range rows {
		var r trackRecord
		var v dbTrackStreamValue
		json.Unmarshal([]byte(row), &r)
		json.Unmarshal([]byte(row), &v)
		key := makeDomesticTrackKey(r)

		if r.CountryCode == r.ReportCountryCode {
			newItemValue, err := json.Marshal(v)
			if err != nil {
				log.Fatal(err)
			}
			item := &pb.BatchPutItem{Key: []byte(key), Value: newItemValue}
			list = append(list, item)
		}
	}

	if len(list) > 0 {
		batchIngest(client, list)
	}

	if err := putIngestItem(client, ingestKey); err != nil {
		log.Printf("processTrackStreamsDomestic: %s", err)
	}
}

func processTrackStreamsRoaming(s3Service *s3.S3, client pb.LevelDBClient, date, licensor, reportCountryCode, s3TargetBucket string) {
	ingestKey := makeIngestTrackStreamsRoamingDBKey(date, reportCountryCode, licensor)
	if isIngested(client, ingestKey) == true {
		log.Println("track_streams_roaming - already ingested", date, licensor, reportCountryCode)
		return
	}

	sourceKey := makeSpotifyTargetStreamsTracksS3Key(date, licensor, reportCountryCode)
	if ok, err := targetFileExists(s3Service, s3TargetBucket, sourceKey); err != nil {
		log.Printf("processTrackStreamsRoaming, error: %s", err)
		return
	} else if !ok {
		log.Println("processTrackStreamsRoaming, track_streams_roaming - source doesn't exist", date, licensor, reportCountryCode, sourceKey)
		return
	}

	log.Println("track_streams_roaming - starting", date, licensor, reportCountryCode, sourceKey)

	rows, err := getRows(s3Service, s3TargetBucket, sourceKey)
	if err != nil {
		log.Fatal(err)
	}

	var list []*pb.BatchPutItem

	for _, row := range rows {
		var r trackRecord
		var v dbTrackStreamValue
		json.Unmarshal([]byte(row), &r)
		json.Unmarshal([]byte(row), &v)
		key := makeRoamingTrackKey(r)

		if r.CountryCode != r.ReportCountryCode {
			item := reduceTrackStreamItem(client, key, v)
			list = append(list, item)
		}
	}

	if len(list) > 0 {
		batchIngest(client, list)
	}

	if err := putIngestItem(client, ingestKey); err != nil {
		log.Printf("processTrackStreamsRoaming: %s", err)
	}
}

func reduceTrackStreamItem(client pb.LevelDBClient, key string, roamingValue dbTrackStreamValue) *pb.BatchPutItem {
	var newValue dbTrackStreamValue
	var prevValue dbTrackStreamValue
	itemKey := []byte(key)

	prevItem, err := client.Get(context.Background(), &pb.GetRequest{Key: itemKey})
	if err != nil && grpc.Code(err) != grpcCodes.NotFound {
		log.Fatal(err)
	}

	if prevItem == nil {
		newValue = roamingValue
	} else {
		json.Unmarshal(prevItem.Value, &prevValue)
		newValue = dbTrackStreamValue{
			PaidListeners:            prevValue.PaidListeners + roamingValue.PaidListeners,
			Listeners:                prevValue.Listeners + roamingValue.Listeners,
			RepeatPlay:               prevValue.RepeatPlay + roamingValue.RepeatPlay,
			ShufflePlay:              prevValue.ShufflePlay + roamingValue.ShufflePlay,
			PaidSourceAlbum:          prevValue.PaidSourceAlbum + roamingValue.PaidSourceAlbum,
			PaidSourceArtist:         prevValue.PaidSourceArtist + roamingValue.PaidSourceArtist,
			PaidSourceChart:          prevValue.PaidSourceChart + roamingValue.PaidSourceChart,
			PaidSourceCollection:     prevValue.PaidSourceCollection + roamingValue.PaidSourceCollection,
			PaidSourceDailyMix:       prevValue.PaidSourceDailyMix + roamingValue.PaidSourceDailyMix,
			PaidSourceDiscoverWeekly: prevValue.PaidSourceDiscoverWeekly + roamingValue.PaidSourceDiscoverWeekly,
			PaidSourceOther:          prevValue.PaidSourceOther + roamingValue.PaidSourceOther,
			PaidSourceOthersPlaylist: prevValue.PaidSourceOthersPlaylist + roamingValue.PaidSourceOthersPlaylist,
			PaidSourcePlayQueue:      prevValue.PaidSourcePlayQueue + roamingValue.PaidSourcePlayQueue,
			PaidSourceRadio:          prevValue.PaidSourceRadio + roamingValue.PaidSourceRadio,
			PaidSourceReleaseRadar:   prevValue.PaidSourceReleaseRadar + roamingValue.PaidSourceReleaseRadar,
			PaidSourceSearch:         prevValue.PaidSourceSearch + roamingValue.PaidSourceSearch,
			PaidSourceUnknown:        prevValue.PaidSourceUnknown + roamingValue.PaidSourceUnknown,
			SourceAlbum:              prevValue.SourceAlbum + roamingValue.SourceAlbum,
			SourceArtist:             prevValue.SourceArtist + roamingValue.SourceArtist,
			SourceChart:              prevValue.SourceChart + roamingValue.SourceChart,
			SourceCollection:         prevValue.SourceCollection + roamingValue.SourceCollection,
			SourceDailyMix:           prevValue.SourceDailyMix + roamingValue.SourceDailyMix,
			SourceDiscoverWeekly:     prevValue.SourceDiscoverWeekly + roamingValue.SourceDiscoverWeekly,
			SourceOther:              prevValue.SourceOther + roamingValue.SourceOther,
			SourceOthersPlaylist:     prevValue.SourceOthersPlaylist + roamingValue.SourceOthersPlaylist,
			SourcePlayQueue:          prevValue.SourcePlayQueue + roamingValue.SourcePlayQueue,
			SourceRadio:              prevValue.SourceRadio + roamingValue.SourceRadio,
			SourceReleaseRadar:       prevValue.SourceReleaseRadar + roamingValue.SourceReleaseRadar,
			SourceSearch:             prevValue.SourceSearch + roamingValue.SourceSearch,
			SourceUnknown:            prevValue.SourceUnknown + roamingValue.SourceUnknown,
			PaidStreams:              prevValue.PaidStreams + roamingValue.PaidStreams,
			Streams:                  prevValue.Streams + roamingValue.Streams,
		}
	}

	newItemValue, err := json.Marshal(newValue)
	if err != nil {
		log.Fatal(err)
	}
	return &pb.BatchPutItem{Key: itemKey, Value: newItemValue}
}

func processTrackPlaylistStreamsDomestic(s3Service *s3.S3, client pb.LevelDBClient, date, licensor, reportCountryCode, s3TargetBucket string) {
	ingestKey := makeIngestTrackPlaylistStreamsDomesticDBKey(date, reportCountryCode, licensor)
	if isIngested(client, ingestKey) == true {
		log.Println("track_playlist_streams_domestic - already ingested", date, licensor, reportCountryCode)
		return
	}

	sourceKey := makeSpotifyTargetStreamsPlaylistTracksS3Key(date, licensor, reportCountryCode)
	if ok, err := targetFileExists(s3Service, s3TargetBucket, sourceKey); err != nil {
		log.Printf("processTrackPlaylistStreamsDomestic, error: %s", err)
		return
	} else if !ok {
		log.Println("processTrackPlaylistStreamsDomestic, track_playlist_streams_domestic - source doesn't exist", date, licensor, reportCountryCode, sourceKey)
		return
	}

	log.Println("track_playlist_streams_domestic - starting", date, licensor, reportCountryCode, sourceKey)

	rows, err := getRows(s3Service, s3TargetBucket, sourceKey)
	if err != nil {
		log.Fatal(err)
	}

	var list []*pb.BatchPutItem
	items := make(map[string]map[string]dbTrackPlaylistStreamValue)

	for _, row := range rows {
		var r playlistTrackRecord
		json.Unmarshal([]byte(row), &r)
		key := makeDomesticTrackPlaylistsKey(r)

		if r.CountryCode == r.ReportCountryCode {
			item, ok := items[key]
			if ok == false {
				item = make(map[string]dbTrackPlaylistStreamValue)
			}

			trackPlaylist, ok := item[r.PlaylistURI]
			if ok == false {
				trackPlaylist = dbTrackPlaylistStreamValue{
					PaidListeners: r.PaidListeners,
					Listeners:     r.Listeners,
					RepeatPlay:    r.RepeatPlay,
					ShufflePlay:   r.ShufflePlay,
					PaidStreams:   r.PaidStreams,
					Streams:       r.Streams,
				}
			} else {
				trackPlaylist = dbTrackPlaylistStreamValue{
					PaidListeners: r.PaidListeners + trackPlaylist.PaidListeners,
					Listeners:     r.Listeners + trackPlaylist.Listeners,
					RepeatPlay:    r.RepeatPlay + trackPlaylist.RepeatPlay,
					ShufflePlay:   r.ShufflePlay + trackPlaylist.ShufflePlay,
					PaidStreams:   r.PaidStreams + trackPlaylist.PaidStreams,
					Streams:       r.Streams + trackPlaylist.Streams,
				}
			}

			item[r.PlaylistURI] = trackPlaylist
			items[key] = item
		}
	}

	for key, v := range items {
		newItemValue, err := json.Marshal(v)
		if err != nil {
			log.Fatal(err)
		}
		item := &pb.BatchPutItem{Key: []byte(key), Value: newItemValue}
		list = append(list, item)
	}

	if len(list) > 0 {
		batchIngest(client, list)
	}

	if err := putIngestItem(client, ingestKey); err != nil {
		log.Printf("processTrackPlaylistStreamsDomestic: %s", err)
	}
}

func processTrackPlaylistStreamsRoaming(s3Service *s3.S3, client pb.LevelDBClient, date, licensor, reportCountryCode, s3TargetBucket string) {
	ingestKey := makeIngestTrackPlaylistStreamsRoamingDBKey(date, reportCountryCode, licensor)
	if isIngested(client, ingestKey) == true {
		log.Println("track_playlist_streams_roaming - already ingested", date, licensor, reportCountryCode)
		return
	}

	sourceKey := makeSpotifyTargetStreamsPlaylistTracksS3Key(date, licensor, reportCountryCode)
	if ok, err := targetFileExists(s3Service, s3TargetBucket, sourceKey); err != nil {
		log.Printf("processTrackPlaylistStreamsRoaming, makeSpotifyTargetStreamsPlaylistTracksS3Key, error: %s", err)
		return
	} else if !ok {
		log.Printf("processTrackPlaylistStreamsRoaming, track_playlist_streams_roaming - source doesn't exist: %s, %s, %s, %s", date, licensor, reportCountryCode, sourceKey)
		return
	}

	log.Println("track_playlist_streams_roaming - starting", date, licensor, reportCountryCode, sourceKey)

	rows, err := getRows(s3Service, s3TargetBucket, sourceKey)
	if err != nil {
		log.Fatal(err)
	}

	var list []*pb.BatchPutItem
	items := make(map[string]map[string]dbTrackPlaylistStreamValue)

	for _, row := range rows {
		var r playlistTrackRecord
		json.Unmarshal([]byte(row), &r)
		key := makeRoamingTrackPlaylistsKey(r)

		if r.CountryCode != r.ReportCountryCode {
			item, ok := items[key]
			if ok == false {
				item = make(map[string]dbTrackPlaylistStreamValue)
			}

			trackPlaylist, ok := item[r.PlaylistURI]
			if ok == false {
				trackPlaylist = dbTrackPlaylistStreamValue{
					PaidListeners: r.PaidListeners,
					Listeners:     r.Listeners,
					RepeatPlay:    r.RepeatPlay,
					ShufflePlay:   r.ShufflePlay,
					PaidStreams:   r.PaidStreams,
					Streams:       r.Streams,
				}
			} else {
				trackPlaylist = dbTrackPlaylistStreamValue{
					PaidListeners: r.PaidListeners + trackPlaylist.PaidListeners,
					Listeners:     r.Listeners + trackPlaylist.Listeners,
					RepeatPlay:    r.RepeatPlay + trackPlaylist.RepeatPlay,
					ShufflePlay:   r.ShufflePlay + trackPlaylist.ShufflePlay,
					PaidStreams:   r.PaidStreams + trackPlaylist.PaidStreams,
					Streams:       r.Streams + trackPlaylist.Streams,
				}
			}

			item[r.PlaylistURI] = trackPlaylist
			items[key] = item
		}
	}

	for key, v := range items {
		item := reduceTrackPlaylistItem(client, key, v)
		list = append(list, item)
	}

	if len(list) > 0 {
		batchIngest(client, list)
	}

	if err := putIngestItem(client, ingestKey); err != nil {
		log.Printf("processTrackPlaylistStreamsRoaming: %s", err)
	}
}

func reduceTrackPlaylistItem(client pb.LevelDBClient, key string, roamingValue map[string]dbTrackPlaylistStreamValue) *pb.BatchPutItem {
	var prevValue map[string]dbTrackPlaylistStreamValue
	newValue := make(map[string]dbTrackPlaylistStreamValue)
	itemKey := []byte(key)

	prevItem, err := client.Get(context.Background(), &pb.GetRequest{Key: itemKey})
	if err != nil && grpc.Code(err) != grpcCodes.NotFound {
		log.Fatal(err)
	}

	if prevItem != nil {
		json.Unmarshal(prevItem.Value, &prevValue)

		for k, v := range prevValue {
			newValue[k] = v
		}
	}

	for k, v := range roamingValue {
		prevItem, ok := newValue[k]
		if ok == false {
			newValue[k] = v
		} else {
			newValue[k] = dbTrackPlaylistStreamValue{
				PaidListeners: prevItem.PaidListeners + v.PaidListeners,
				Listeners:     prevItem.Listeners + v.Listeners,
				RepeatPlay:    prevItem.RepeatPlay + v.RepeatPlay,
				ShufflePlay:   prevItem.ShufflePlay + v.ShufflePlay,
				PaidStreams:   prevItem.PaidStreams + v.PaidStreams,
				Streams:       prevItem.Streams + v.Streams,
			}
		}
	}

	newItemValue, err := json.Marshal(newValue)
	if err != nil {
		log.Fatal(err)
	}
	return &pb.BatchPutItem{Key: itemKey, Value: newItemValue}
}

func batchIngest(client pb.LevelDBClient, list []*pb.BatchPutItem) {
	chunks := int(math.Ceil(float64(len(list)) / 100))
	count := 0

	for i := 0; i < chunks; i++ {
		newCount := count + 100
		start := int(math.Min(float64(count), float64(len(list))))
		end := int(math.Min(float64(newCount), float64(len(list))))
		items := list[start:end]

		r, err := client.BatchPut(context.Background(), &pb.BatchPutRequest{Items: items})
		if err != nil {
			log.Fatal(err)
		}
		_ = r
		count = newCount
	}
}

func putIngestItem(client pb.LevelDBClient, key []byte) error {
	iv := ingestValue{
		Timestamp: time.Now().UTC().Format(time.RFC3339Nano),
	}
	jsonIv, err1 := json.Marshal(iv)
	if err1 != nil {
		return fmt.Errorf("putIngestItem: %s", err1)
	}
	_, err2 := client.Put(context.Background(), &pb.PutRequest{Key: key, Value: jsonIv})
	if err2 != nil {
		return fmt.Errorf("putIngestItem: %s", err2)
	}
	return nil
}

func makeDomesticTrackKey(r trackRecord) string {
	if r.Date >= "2018-01-22" {
		return "Aÿ" + r.ISRC + "ÿ" + r.Date + "ÿ" + r.CountryCode + "ÿ" + r.Licensor + "ÿdomestic"
	}
	return "Aÿ" + r.ISRC + "ÿ" + r.Date + "ÿ" + r.CountryCode + "ÿ" + r.Licensor
}

func makeRoamingTrackKey(r trackRecord) string {
	return "Aÿ" + r.ISRC + "ÿ" + r.Date + "ÿ" + r.CountryCode + "ÿ" + r.Licensor + "ÿroaming"
}

func makeDomesticTrackPlaylistsKey(r playlistTrackRecord) string {
	if r.Date >= "2018-01-22" {
		return "Bÿ" + r.ISRC + "ÿ" + r.Date + "ÿ" + r.CountryCode + "ÿ" + r.Licensor + "ÿdomestic"
	}
	return "Bÿ" + r.ISRC + "ÿ" + r.Date + "ÿ" + r.CountryCode + "ÿ" + r.Licensor
}

func makeRoamingTrackPlaylistsKey(r playlistTrackRecord) string {
	return "Bÿ" + r.ISRC + "ÿ" + r.Date + "ÿ" + r.CountryCode + "ÿ" + r.Licensor + "ÿroaming"
}

func makeIngestTrackStreamsDomesticDBKey(date string, reportCountryCode string, licensor string) []byte {
	return []byte("Zÿtrack_streams_domesticÿ" + date + "ÿ" + reportCountryCode + "ÿ" + licensor)
}

func makeIngestTrackStreamsRoamingDBKey(date string, reportCountryCode string, licensor string) []byte {
	return []byte("Zÿtrack_streams_roamingÿ" + date + "ÿ" + reportCountryCode + "ÿ" + licensor)
}

func makeIngestTrackPlaylistStreamsDomesticDBKey(date string, reportCountryCode string, licensor string) []byte {
	return []byte("Zÿtrack_playlist_streams_domesticÿ" + date + "ÿ" + reportCountryCode + "ÿ" + licensor)
}

func makeIngestTrackPlaylistStreamsRoamingDBKey(date string, reportCountryCode string, licensor string) []byte {
	return []byte("Zÿtrack_playlist_streams_roamingÿ" + date + "ÿ" + reportCountryCode + "ÿ" + licensor)
}

func isIngested(client pb.LevelDBClient, key []byte) bool {
	_, err := client.Get(context.Background(), &pb.GetRequest{Key: key})
	return err == nil
}

// IngestStreams ingests Spotify streams
func IngestStreams(s3Service *s3.S3, startDate time.Time, endDate time.Time, licensors []string, dbAddress, s3Bucket string) {
	conn, err := grpc.Dial(dbAddress, grpc.WithInsecure())
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close()
	client := pb.NewLevelDBClient(conn)
	dates := daterange.DateRange(startDate, endDate)

	for _, date := range dates {
		for _, licensor := range licensors {
			if date >= "2018-01-22" {
				for _, countryCode := range V2CountryCodes {
					processTrackStreamsDomestic(s3Service, client, date, licensor, countryCode, s3Bucket)
					processTrackPlaylistStreamsDomestic(s3Service, client, date, licensor, countryCode, s3Bucket)
					processTrackStreamsRoaming(s3Service, client, date, licensor, countryCode, s3Bucket)
					processTrackPlaylistStreamsRoaming(s3Service, client, date, licensor, countryCode, s3Bucket)
				}
			} else {
				for _, countryCode := range V1CountryCodes {
					processTrackStreamsDomestic(s3Service, client, date, licensor, countryCode, s3Bucket)
					processTrackPlaylistStreamsDomestic(s3Service, client, date, licensor, countryCode, s3Bucket)
				}
			}
		}
	}
}