package spotifyconsumeranalytics

import (
	"bufio"
	"compress/gzip"
	"encoding/json"
	"io"
	"log"
	"regexp"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
	"github.com/filtr/go-apollo/pkg/daterange"
)

type album struct {
	Name string `json:"name"`
	URI  string `json:"href"`
}

type track struct {
	ISRC string `json:"isrc"`
	Name string `json:"name"`
	URI  string `json:"href"`
}

type trackv2 struct {
	ISRC string `json:"isrc"`
	Name string `json:"name"`
	URI  string `json:"href"`
}

type artists struct {
	Names string `json:"names"`
	URIs  string `json:"hrefs"`
}

type savesProduct struct {
	Free    int `json:"free"`
	Premium int `json:"premium"`
}

type saveCountry struct {
	Total   int          `json:"total"`
	Product savesProduct `json:"product"`
}

type skipCountry struct {
	Total int `json:"total"`
}

type saves struct {
	Total   int                    `json:"total"`
	Country map[string]saveCountry `json:"country"`
}

type skips struct {
	Total   int                    `json:"total"`
	Country map[string]skipCountry `json:"country"`
}

type streamGenderAge struct {
	Age map[string]int `json:"age"`
}

type streamGender struct {
	Male    streamGenderAge `json:"male"`
	Female  streamGenderAge `json:"female"`
	Neutral streamGenderAge `json:"neutral"`
	Unknown streamGenderAge `json:"unknown"`
}

type streamCountry struct {
	Total  int          `json:"total"`
	Gender streamGender `json:"sex"`
}

type streams struct {
	Total   int                      `json:"total"`
	Country map[string]streamCountry `json:"country"`
}

type apiRecord struct {
	Album        album   `json:"album"`
	Artists      artists `json:"artists"`
	Track        track   `json:"track"`
	Trackv2      trackv2 `json:"trackv2"`
	Date         string  `json:"date"`
	LicensorName string  `json:"licensor"`
	Label        string  `json:"label"`
	Saves        saves   `json:"saves"`
	Skips        skips   `json:"skips"`
	Streams      streams `json:"streams"`
}

type demographicsRecord struct {
	ReportDate              string `json:"report_date"`
	ReportCountryCode       string `json:"report_country_code"`
	Date                    string `json:"date"`
	CountryCode             string `json:"country_code"`
	Licensor                string `json:"licensor"`
	ISRC                    string `json:"isrc"`
	Streams                 int    `json:"streams"`
	GenderFemaleAge0_17     int    `json:"gender_female_age_0_17"`
	GenderMaleAge0_17       int    `json:"gender_male_age_0_17"`
	GenderNeutralAge0_17    int    `json:"gender_neutral_age_0_17"`
	GenderUnknownAge0_17    int    `json:"gender_unknown_age_0_17"`
	GenderFemaleAge18_22    int    `json:"gender_female_age_18_22"`
	GenderMaleAge18_22      int    `json:"gender_male_age_18_22"`
	GenderNeutralAge18_22   int    `json:"gender_neutral_age_18_22"`
	GenderUnknownAge18_22   int    `json:"gender_unknown_age_18_22"`
	GenderFemaleAge23_27    int    `json:"gender_female_age_23_27"`
	GenderMaleAge23_27      int    `json:"gender_male_age_23_27"`
	GenderNeutralAge23_27   int    `json:"gender_neutral_age_23_27"`
	GenderUnknownAge23_27   int    `json:"gender_unknown_age_23_27"`
	GenderFemaleAge28_34    int    `json:"gender_female_age_28_34"`
	GenderMaleAge28_34      int    `json:"gender_male_age_28_34"`
	GenderNeutralAge28_34   int    `json:"gender_neutral_age_28_34"`
	GenderUnknownAge28_34   int    `json:"gender_unknown_age_28_34"`
	GenderFemaleAge35_44    int    `json:"gender_female_age_35_44"`
	GenderMaleAge35_44      int    `json:"gender_male_age_35_44"`
	GenderNeutralAge35_44   int    `json:"gender_neutral_age_35_44"`
	GenderUnknownAge35_44   int    `json:"gender_unknown_age_35_44"`
	GenderFemaleAge45_59    int    `json:"gender_female_age_45_59"`
	GenderMaleAge45_59      int    `json:"gender_male_age_45_59"`
	GenderNeutralAge45_59   int    `json:"gender_neutral_age_45_59"`
	GenderUnknownAge45_59   int    `json:"gender_unknown_age_45_59"`
	GenderFemaleAge60_150   int    `json:"gender_female_age_60_150"`
	GenderMaleAge60_150     int    `json:"gender_male_age_60_150"`
	GenderNeutralAge60_150  int    `json:"gender_neutral_age_60_150"`
	GenderUnknownAge60_150  int    `json:"gender_unknown_age_60_150"`
	GenderFemaleAgeUnknown  int    `json:"gender_female_age_unknown"`
	GenderMaleAgeUnknown    int    `json:"gender_male_age_unknown"`
	GenderNeutralAgeUnknown int    `json:"gender_neutral_age_unknown"`
	GenderUnknownAgeUnknown int    `json:"gender_unknown_age_unknown"`
}

type savesSkipsRecord struct {
	ReportDate        string `json:"report_date"`
	ReportCountryCode string `json:"report_country_code"`
	Date              string `json:"date"`
	CountryCode       string `json:"country_code"`
	Licensor          string `json:"licensor"`
	ISRC              string `json:"isrc"`
	Saves             int    `json:"saves"`
	FreeSaves         int    `json:"free_saves"`
	PaidSaves         int    `json:"paid_saves"`
	Skips             int    `json:"skips"`
	Streams           int    `json:"streams"`
}

var reTrackName = regexp.MustCompile(`"trackv2": {"name": "...+", "href": "spotify:track:`)
var reAlbumName = regexp.MustCompile(`"album": {"name": "...+", "href": "spotify:album:`)
var reArtistsNames = regexp.MustCompile(`"artists": {"names": "...+", "hrefs": "spotify:artist:`)
var reLabel = regexp.MustCompile(`"label": "...+"}`)

func fixJSON(s string) string {
	s1 := reTrackName.ReplaceAllString(s, `"trackv2": {"name": "", "href": "spotify:track:`)
	s2 := reAlbumName.ReplaceAllString(s1, `"album": {"name": "", "href": "spotify:album:`)
	s3 := reArtistsNames.ReplaceAllString(s2, `"artists": {"names": "", "hrefs": "spotify:artist:`)
	s4 := reLabel.ReplaceAllString(s3, `"label": ""}`)
	return s4
}

func processAggregatedStreams(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket, date, licensor string, tracks map[string]string) {
	input := &s3.GetObjectInput{
		Bucket: aws.String(s3SourceBucket),
		Key:    aws.String(makeSourceAggregatedStreamsS3Key(date, licensor)),
	}

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

	savesSkipsMap := make(map[string]savesSkipsRecord)
	demographicsMap := make(map[string]demographicsRecord)

	reader, err := gzip.NewReader(result.Body)
	if err != nil {
		log.Fatal(err)
	}
	defer reader.Close()

	bufReader := bufio.NewReader(reader)
	for {
		line, err := bufReader.ReadString('\n')
		if err != nil {
			if err == io.EOF {
				break
			}

			log.Fatal(err)
		}

		var record apiRecord

		err2 := json.Unmarshal([]byte(line), &record)
		if err2 != nil {
			err3 := json.Unmarshal([]byte(fixJSON(line)), &record)
			if err3 != nil {
				log.Fatal(err3)
			}
		}

		var uri string
		var isrc string
		var d string
		if record.Date == "" {
			d = date
		} else {
			d = record.Date
		}

		if record.Trackv2.URI != "" {
			uri = record.Trackv2.URI
		} else if record.Track.URI != "" {
			uri = record.Track.URI
		}

		if record.Trackv2.ISRC != "" {
			isrc = record.Trackv2.ISRC
		} else if record.Track.ISRC != "" {
			isrc = record.Track.ISRC
		} else {
			v := tracks[uri]
			if v == "" {
				continue
			}
			isrc = v
		}

		if uri == "" && isrc == "" {
			continue
		}

		if isrc == "" {
			log.Println("no isrc")
		}

		for c := range record.Saves.Country {
			countryCode := strings.ToLower(c)
			saves := record.Saves.Country[c].Total
			freeSaves := record.Saves.Country[c].Product.Free
			premiumSaves := record.Saves.Country[c].Product.Premium
			key := isrc + countryCode + d
			v, ok := savesSkipsMap[key]
			if ok == false {
				newV := savesSkipsRecord{
					ReportDate:        date,
					ReportCountryCode: countryCode,
					Date:              d,
					CountryCode:       countryCode,
					Licensor:          licensor,
					ISRC:              isrc,
					Saves:             saves,
					FreeSaves:         freeSaves,
					PaidSaves:         premiumSaves,
				}
				savesSkipsMap[key] = newV
			} else {
				v.Saves = v.Saves + saves
				v.FreeSaves = v.FreeSaves + freeSaves
				v.PaidSaves = v.PaidSaves + premiumSaves
				savesSkipsMap[key] = v
			}
		}

		for c := range record.Skips.Country {
			countryCode := strings.ToLower(c)
			skips := record.Skips.Country[c].Total
			key := isrc + countryCode + d
			v, ok := savesSkipsMap[key]
			if ok == false {
				newV := savesSkipsRecord{
					ReportDate:        date,
					ReportCountryCode: countryCode,
					Date:              d,
					CountryCode:       countryCode,
					Licensor:          licensor,
					ISRC:              isrc,
					Skips:             skips,
				}
				savesSkipsMap[key] = newV
			} else {
				v.Skips = v.Skips + skips
				savesSkipsMap[key] = v
			}
		}

		for c := range record.Streams.Country {
			countryCode := strings.ToLower(c)
			streams := record.Streams.Country[c].Total
			key := isrc + countryCode + d
			v, ok := savesSkipsMap[key]
			if ok == false {
				newV := savesSkipsRecord{
					ReportDate:        date,
					ReportCountryCode: countryCode,
					Date:              d,
					CountryCode:       countryCode,
					Licensor:          licensor,
					ISRC:              isrc,
					Streams:           streams,
				}
				savesSkipsMap[key] = newV
			} else {
				v.Streams = v.Streams + streams
				savesSkipsMap[key] = v
			}

			av, ok := demographicsMap[key]
			if ok == false {
				av = demographicsRecord{
					ReportDate:        date,
					ReportCountryCode: countryCode,
					Date:              d,
					CountryCode:       countryCode,
					Licensor:          licensor,
					ISRC:              isrc,
				}
			}

			av.Streams = av.Streams + streams

			for a := range record.Streams.Country[c].Gender.Female.Age {
				s := record.Streams.Country[c].Gender.Female.Age[a]
				switch a {
				case "0-17":
					av.GenderFemaleAge0_17 = av.GenderFemaleAge0_17 + s
				case "18-22":
					av.GenderFemaleAge18_22 = av.GenderFemaleAge18_22 + s
				case "23-27":
					av.GenderFemaleAge23_27 = av.GenderFemaleAge23_27 + s
				case "28-34":
					av.GenderFemaleAge28_34 = av.GenderFemaleAge28_34 + s
				case "35-44":
					av.GenderFemaleAge35_44 = av.GenderFemaleAge35_44 + s
				case "45-59":
					av.GenderFemaleAge45_59 = av.GenderFemaleAge45_59 + s
				case "60-150":
					av.GenderFemaleAge60_150 = av.GenderFemaleAge60_150 + s
				default:
					av.GenderFemaleAgeUnknown = av.GenderFemaleAgeUnknown + s
				}
			}
			for a := range record.Streams.Country[c].Gender.Male.Age {
				s := record.Streams.Country[c].Gender.Male.Age[a]
				switch a {
				case "0-17":
					av.GenderMaleAge0_17 = av.GenderMaleAge0_17 + s
				case "18-22":
					av.GenderMaleAge18_22 = av.GenderMaleAge18_22 + s
				case "23-27":
					av.GenderMaleAge23_27 = av.GenderMaleAge23_27 + s
				case "28-34":
					av.GenderMaleAge28_34 = av.GenderMaleAge28_34 + s
				case "35-44":
					av.GenderMaleAge35_44 = av.GenderMaleAge35_44 + s
				case "45-59":
					av.GenderMaleAge45_59 = av.GenderMaleAge45_59 + s
				case "60-150":
					av.GenderMaleAge60_150 = av.GenderMaleAge60_150 + s
				default:
					av.GenderMaleAgeUnknown = av.GenderMaleAgeUnknown + s
				}
			}
			for a := range record.Streams.Country[c].Gender.Neutral.Age {
				s := record.Streams.Country[c].Gender.Neutral.Age[a]
				switch a {
				case "0-17":
					av.GenderNeutralAge0_17 = av.GenderNeutralAge0_17 + s
				case "18-22":
					av.GenderNeutralAge18_22 = av.GenderNeutralAge18_22 + s
				case "23-27":
					av.GenderNeutralAge23_27 = av.GenderNeutralAge23_27 + s
				case "28-34":
					av.GenderNeutralAge28_34 = av.GenderNeutralAge28_34 + s
				case "35-44":
					av.GenderNeutralAge35_44 = av.GenderNeutralAge35_44 + s
				case "45-59":
					av.GenderNeutralAge45_59 = av.GenderNeutralAge45_59 + s
				case "60-150":
					av.GenderNeutralAge60_150 = av.GenderNeutralAge60_150 + s
				default:
					av.GenderNeutralAgeUnknown = av.GenderNeutralAgeUnknown + s
				}
			}
			for a := range record.Streams.Country[c].Gender.Unknown.Age {
				s := record.Streams.Country[c].Gender.Unknown.Age[a]
				switch a {
				case "0-17":
					av.GenderUnknownAge0_17 = av.GenderUnknownAge0_17 + s
				case "18-22":
					av.GenderUnknownAge18_22 = av.GenderUnknownAge18_22 + s
				case "23-27":
					av.GenderUnknownAge23_27 = av.GenderUnknownAge23_27 + s
				case "28-34":
					av.GenderUnknownAge28_34 = av.GenderUnknownAge28_34 + s
				case "35-44":
					av.GenderUnknownAge35_44 = av.GenderUnknownAge35_44 + s
				case "45-59":
					av.GenderUnknownAge45_59 = av.GenderUnknownAge45_59 + s
				case "60-150":
					av.GenderUnknownAge60_150 = av.GenderUnknownAge60_150 + s
				default:
					av.GenderUnknownAgeUnknown = av.GenderUnknownAgeUnknown + s
				}
			}

			demographicsMap[key] = av
		}
	}

	savesSkipsSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)}))
	savesSkipsUploader := s3manager.NewUploader(savesSkipsSession)
	savesSkipsReader, savesSkipsWriter := io.Pipe()

	demographicsSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)}))
	demographicsUploader := s3manager.NewUploader(demographicsSession)
	demographicsReader, demographicsWriter := io.Pipe()

	go func() {
		gw := gzip.NewWriter(savesSkipsWriter)
		for _, v := range savesSkipsMap {
			row, err := json.Marshal(v)
			if err != nil {
				log.Fatalf("processAggregatedStreams, savesSkipsWriter: %s", err)
			}

			r := append(row, []byte("\n")...)

			if _, err := gw.Write(r); err != nil {
				log.Fatalf("processAggregatedStreams, savesSkipsWriter: %s", err)
			}
		}
		if err := gw.Close(); err != nil {
			log.Fatalf("processAggregatedStreams, savesSkipsWriter: %s", err)
		}
		if err := savesSkipsWriter.Close(); err != nil {
			log.Fatalf("processAggregatedStreams, savesSkipsWriter: %s", err)
		}
	}()

	go func() {
		gw := gzip.NewWriter(demographicsWriter)
		for _, v := range demographicsMap {
			row, err := json.Marshal(v)
			if err != nil {
				log.Fatalf("processAggregatedStreams, demographicsWriter: %s", err)
			}

			r := append(row, []byte("\n")...)

			if _, err := gw.Write(r); err != nil {
				log.Fatalf("processAggregatedStreams, demographicsWriter: %s", err)
			}
		}
		if err := gw.Close(); err != nil {
			log.Fatalf("processAggregatedStreams, demographicsWriter: %s", err)
		}
		if err := demographicsWriter.Close(); err != nil {
			log.Fatalf("processAggregatedStreams, demographicsWriter: %s", err)
		}
	}()

	if _, err := savesSkipsUploader.Upload(&s3manager.UploadInput{
		Bucket:          aws.String(s3TargetBucket),
		Key:             aws.String(makeTargetSavesSkipsTracksS3Key(date, licensor)),
		Body:            savesSkipsReader,
		ContentType:     aws.String("application/x-ndjson"),
		ContentEncoding: aws.String("gzip"),
		ACL:             aws.String("bucket-owner-full-control"),
	}); err != nil {
		log.Fatalf("processAggregatedStreams, savesSkipsUploader: %s", err)
	}

	if _, err := demographicsUploader.Upload(&s3manager.UploadInput{
		Bucket:          aws.String(s3TargetBucket),
		Key:             aws.String(makeTargetDemographicsTracksS3Key(date, licensor)),
		Body:            demographicsReader,
		ContentType:     aws.String("application/x-ndjson"),
		ContentEncoding: aws.String("gzip"),
		ACL:             aws.String("bucket-owner-full-control"),
	}); err != nil {
		log.Fatalf("processAggregatedStreams, demographicsUploader: %s", err)
	}
}

func runAggregatedStreamsTask(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket, date, licensor string) {
	demographicsExist, err := targetFileExists(s3TargetService, s3TargetBucket, makeTargetDemographicsTracksS3Key(date, licensor))
	if err != nil {
		log.Printf("runAggregatedStreamsTask, error: %s", err)
	}
	savesSkipsExist, err := targetFileExists(s3TargetService, s3TargetBucket, makeTargetSavesSkipsTracksS3Key(date, licensor))
	if err != nil {
		log.Printf("runAggregatedStreamsTask, error: %s", err)
	}

	if demographicsExist && savesSkipsExist {
		log.Println("runAggregatedStreamsTask, already processed", date, licensor)
		return
	}

	if ok, err := sourceAggregatedStreamsFileExists(s3SourceService, s3SourceBucket, date, licensor); err != nil {
		log.Printf("runAggregatedStreamsTask, sourceAggregatedStreamsFileExists, error: %s", err)
		return
	} else if !ok {
		log.Printf("runAggregatedStreamsTask, aggregatedstreams doesn't exist: %s %s", date, licensor)
		return
	}

	if ok, err := sourceSpotifyTracksFileExists(s3SourceService, s3SourceBucket, date, licensor); err != nil {
		log.Printf("runAggregatedStreamsTask, sourceSpotifyTracksFileExists: %s", err)
		return
	} else if !ok {
		log.Printf("runAggregatedStreamsTask, tracks doesn't exist: %s %s", date, licensor)
		return
	}

	log.Println("fetching tracks", date, licensor)
	tracks := getTracksMapURIToISRC(s3SourceService, s3SourceBucket, date, licensor)

	log.Println("starting", date, licensor)
	processAggregatedStreams(s3SourceService, s3TargetService, s3SourceBucket, s3TargetBucket, date, licensor, tracks)
}

// RunAggregatedStreams aggregates saves, skips and demographics
func RunAggregatedStreams(s3SourceService *s3.S3, s3TargetService *s3.S3, startDate time.Time, endDate time.Time, licensors []string, s3SourceBucket, s3TargetBucket string) {
	dates := daterange.DateRange(startDate, endDate)

	for _, date := range dates {
		for _, licensor := range licensors {
			runAggregatedStreamsTask(s3SourceService, s3TargetService, s3SourceBucket, s3TargetBucket, date, licensor)
		}
	}
}