package main

import (
	"bufio"
	"compress/gzip"
	"encoding/json"
	"flag"
	"io"
	"log"
	"sort"
	"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"
)

var inputStartDate string
var inputEndDate string

func init() {
	defaultStartDate := time.Now().AddDate(0, 0, -11).Format("2006-01-02")
	defaultEndDate := time.Now().AddDate(0, 0, -1).Format("2006-01-02")
	flag.StringVar(&inputStartDate, "startdate", defaultStartDate, "StartDate")
	flag.StringVar(&inputEndDate, "enddate", defaultEndDate, "EndDate")
}

type trackRecord struct {
	ReportDate                      string `json:"report_date"`
	ReportCountryCode               string `json:"report_country_code"`
	Date                            string `json:"date"`
	CountryCode                     string `json:"country_code"`
	ISRC                            string `json:"isrc"`
	DeviceTypePersonalComputer      int    `json:"device_type_personal_computer"`
	DeviceTypeCellPhone             int    `json:"device_type_cell_phone"`
	DeviceTypeTablet                int    `json:"device_type_tablet"`
	DeviceTypeGamingConsole         int    `json:"device_type_gaming_console"`
	DeviceTypeSmartTVDevice         int    `json:"device_type_smart_tv_device"`
	DeviceTypeConnectedAudioDevice  int    `json:"device_type_connected_audio_device"`
	DeviceTypeBuiltinCarApplication int    `json:"device_type_builtin_car_application"`
	DeviceTypeOther                 int    `json:"device_type_other"`
	DeviceOSAndroid                 int    `json:"device_os_android"`
	DeviceOSBlackBerry              int    `json:"device_os_blackberry"`
	DeviceOSBrowser                 int    `json:"device_os_browser"`
	DeviceOSiOS                     int    `json:"device_os_ios"`
	DeviceOSLinux                   int    `json:"device_os_linux"`
	DeviceOSMac                     int    `json:"device_os_mac"`
	DeviceOSOther                   int    `json:"device_os_other"`
	DeviceOSWindows                 int    `json:"device_os_windows"`
	RepeatPlay                      int    `json:"repeat_play"`
	ShufflePlay                     int    `json:"shuffle_play"`
	SourceAlbum                     int    `json:"source_album"`
	SourceArtist                    int    `json:"source_artist"`
	SourceChart                     int    `json:"source_chart"`
	SourceCollection                int    `json:"source_collection"`
	SourceOther                     int    `json:"source_other"`
	SourceOthersPlaylist            int    `json:"source_others_playlist"`
	SourceRadio                     int    `json:"source_radio"`
	SourceSearch                    int    `json:"source_search"`
	Streams                         int    `json:"streams"`
}

type summaryRecord struct {
	ReportDate                      string `json:"report_date"`
	ReportCountryCode               string `json:"report_country_code"`
	Date                            string `json:"date"`
	CountryCode                     string `json:"country_code"`
	DeviceTypePersonalComputer      int    `json:"device_type_personal_computer"`
	DeviceTypeCellPhone             int    `json:"device_type_cell_phone"`
	DeviceTypeTablet                int    `json:"device_type_tablet"`
	DeviceTypeGamingConsole         int    `json:"device_type_gaming_console"`
	DeviceTypeSmartTVDevice         int    `json:"device_type_smart_tv_device"`
	DeviceTypeConnectedAudioDevice  int    `json:"device_type_connected_audio_device"`
	DeviceTypeBuiltinCarApplication int    `json:"device_type_builtin_car_application"`
	DeviceTypeOther                 int    `json:"device_type_other"`
	DeviceOSAndroid                 int    `json:"device_os_android"`
	DeviceOSBlackBerry              int    `json:"device_os_blackberry"`
	DeviceOSBrowser                 int    `json:"device_os_browser"`
	DeviceOSiOS                     int    `json:"device_os_ios"`
	DeviceOSLinux                   int    `json:"device_os_linux"`
	DeviceOSMac                     int    `json:"device_os_mac"`
	DeviceOSOther                   int    `json:"device_os_other"`
	DeviceOSWindows                 int    `json:"device_os_windows"`
	RepeatPlay                      int    `json:"repeat_play"`
	ShufflePlay                     int    `json:"shuffle_play"`
	SourceAlbum                     int    `json:"source_album"`
	SourceArtist                    int    `json:"source_artist"`
	SourceChart                     int    `json:"source_chart"`
	SourceCollection                int    `json:"source_collection"`
	SourceOther                     int    `json:"source_other"`
	SourceOthersPlaylist            int    `json:"source_others_playlist"`
	SourceRadio                     int    `json:"source_radio"`
	SourceSearch                    int    `json:"source_search"`
	Streams                         int    `json:"streams"`
}

type playlistRecord struct {
	ReportDate                      string `json:"report_date"`
	ReportCountryCode               string `json:"report_country_code"`
	Date                            string `json:"date"`
	CountryCode                     string `json:"country_code"`
	PlaylistURI                     string `json:"playlist_uri"`
	DeviceTypePersonalComputer      int    `json:"device_type_personal_computer"`
	DeviceTypeCellPhone             int    `json:"device_type_cell_phone"`
	DeviceTypeTablet                int    `json:"device_type_tablet"`
	DeviceTypeGamingConsole         int    `json:"device_type_gaming_console"`
	DeviceTypeSmartTVDevice         int    `json:"device_type_smart_tv_device"`
	DeviceTypeConnectedAudioDevice  int    `json:"device_type_connected_audio_device"`
	DeviceTypeBuiltinCarApplication int    `json:"device_type_builtin_car_application"`
	DeviceTypeOther                 int    `json:"device_type_other"`
	DeviceOSAndroid                 int    `json:"device_os_android"`
	DeviceOSBlackBerry              int    `json:"device_os_blackberry"`
	DeviceOSBrowser                 int    `json:"device_os_browser"`
	DeviceOSiOS                     int    `json:"device_os_ios"`
	DeviceOSLinux                   int    `json:"device_os_linux"`
	DeviceOSMac                     int    `json:"device_os_mac"`
	DeviceOSOther                   int    `json:"device_os_other"`
	DeviceOSWindows                 int    `json:"device_os_windows"`
	RepeatPlay                      int    `json:"repeat_play"`
	ShufflePlay                     int    `json:"shuffle_play"`
	Streams                         int    `json:"streams"`
}

type playlistTrackRecord struct {
	ReportDate                      string `json:"report_date"`
	ReportCountryCode               string `json:"report_country_code"`
	Date                            string `json:"date"`
	CountryCode                     string `json:"country_code"`
	PlaylistURI                     string `json:"playlist_uri"`
	ISRC                            string `json:"isrc"`
	DeviceTypePersonalComputer      int    `json:"device_type_personal_computer"`
	DeviceTypeCellPhone             int    `json:"device_type_cell_phone"`
	DeviceTypeTablet                int    `json:"device_type_tablet"`
	DeviceTypeGamingConsole         int    `json:"device_type_gaming_console"`
	DeviceTypeSmartTVDevice         int    `json:"device_type_smart_tv_device"`
	DeviceTypeConnectedAudioDevice  int    `json:"device_type_connected_audio_device"`
	DeviceTypeBuiltinCarApplication int    `json:"device_type_builtin_car_application"`
	DeviceTypeOther                 int    `json:"device_type_other"`
	DeviceOSAndroid                 int    `json:"device_os_android"`
	DeviceOSBlackBerry              int    `json:"device_os_blackberry"`
	DeviceOSBrowser                 int    `json:"device_os_browser"`
	DeviceOSiOS                     int    `json:"device_os_ios"`
	DeviceOSLinux                   int    `json:"device_os_linux"`
	DeviceOSMac                     int    `json:"device_os_mac"`
	DeviceOSOther                   int    `json:"device_os_other"`
	DeviceOSWindows                 int    `json:"device_os_windows"`
	RepeatPlay                      int    `json:"repeat_play"`
	ShufflePlay                     int    `json:"shuffle_play"`
	Streams                         int    `json:"streams"`
}

func unmarshalSpotifyStreamRecord(data []byte) spotifyStreamRecord {
	var stream spotifyStreamRecord
	json.Unmarshal(data, &stream)
	stream.DeviceOS = strings.ToLower(stream.DeviceOS)
	if stream.Shuffle != "" && stream.ShufflePlay == "" {
		stream.ShufflePlay = stream.Shuffle
	}
	stream.ShufflePlay = strings.ToLower(stream.ShufflePlay)
	return stream
}

func mutateTrack(t trackRecord, stream spotifyStreamRecord) trackRecord {
	t.Streams++

	switch source := stream.Source; source {
	case "album":
		t.SourceAlbum++
	case "artist":
		t.SourceArtist++
	case "chart":
		t.SourceChart++
	case "collection":
		t.SourceCollection++
	case "other":
		t.SourceOther++
	case "others_playlist":
		t.SourceOthersPlaylist++
	case "radio":
		t.SourceRadio++
	case "search":
		t.SourceSearch++
	}

	if stream.ShufflePlay == "y" {
		t.ShufflePlay++
	}
	if stream.RepeatPlay == "y" {
		t.RepeatPlay++
	}

	switch deviceType := stream.DeviceType; deviceType {
	case "desktop":
		t.DeviceTypePersonalComputer++
	case "mobile":
		t.DeviceTypeCellPhone++
	case "tablet":
		t.DeviceTypeTablet++
	case "personal computer":
		t.DeviceTypePersonalComputer++
	case "cell phone":
		t.DeviceTypeCellPhone++
	case "smart tv device":
		t.DeviceTypeSmartTVDevice++
	case "connected audio device":
		t.DeviceTypeConnectedAudioDevice++
	case "built-in car application":
		t.DeviceTypeBuiltinCarApplication++
	default:
		t.DeviceTypeOther++
	}

	switch deviceOS := stream.DeviceOS; deviceOS {
	case "android":
		t.DeviceOSAndroid++
	case "blackberry":
		t.DeviceOSBlackBerry++
	case "browser":
		t.DeviceOSBrowser++
	case "ios":
		t.DeviceOSiOS++
	case "linux":
		t.DeviceOSLinux++
	case "mac":
		t.DeviceOSMac++
	case "windows":
		t.DeviceOSWindows++
	default:
		t.DeviceOSOther++
	}

	return t
}

func mutateSummary(s summaryRecord, stream spotifyStreamRecord) summaryRecord {
	s.Streams++

	switch source := stream.Source; source {
	case "album":
		s.SourceAlbum++
	case "artist":
		s.SourceArtist++
	case "chart":
		s.SourceChart++
	case "collection":
		s.SourceCollection++
	case "other":
		s.SourceOther++
	case "others_playlist":
		s.SourceOthersPlaylist++
	case "radio":
		s.SourceRadio++
	case "search":
		s.SourceSearch++
	}

	if stream.ShufflePlay == "y" {
		s.ShufflePlay++
	}
	if stream.RepeatPlay == "y" {
		s.RepeatPlay++
	}

	switch deviceType := stream.DeviceType; deviceType {
	case "desktop":
		s.DeviceTypePersonalComputer++
	case "mobile":
		s.DeviceTypeCellPhone++
	case "tablet":
		s.DeviceTypeTablet++
	case "personal computer":
		s.DeviceTypePersonalComputer++
	case "cell phone":
		s.DeviceTypeCellPhone++
	case "smart tv device":
		s.DeviceTypeSmartTVDevice++
	case "connected audio device":
		s.DeviceTypeConnectedAudioDevice++
	case "built-in car application":
		s.DeviceTypeBuiltinCarApplication++
	default:
		s.DeviceTypeOther++
	}

	switch deviceOS := stream.DeviceOS; deviceOS {
	case "android":
		s.DeviceOSAndroid++
	case "blackberry":
		s.DeviceOSBlackBerry++
	case "browser":
		s.DeviceOSBrowser++
	case "ios":
		s.DeviceOSiOS++
	case "linux":
		s.DeviceOSLinux++
	case "mac":
		s.DeviceOSMac++
	case "windows":
		s.DeviceOSWindows++
	default:
		s.DeviceOSOther++
	}

	return s
}

func mutatePlaylist(p playlistRecord, stream spotifyStreamRecord) playlistRecord {
	p.Streams++

	if stream.ShufflePlay == "y" {
		p.ShufflePlay++
	}
	if stream.RepeatPlay == "y" {
		p.RepeatPlay++
	}

	switch deviceType := stream.DeviceType; deviceType {
	case "desktop":
		p.DeviceTypePersonalComputer++
	case "mobile":
		p.DeviceTypeCellPhone++
	case "tablet":
		p.DeviceTypeTablet++
	case "personal computer":
		p.DeviceTypePersonalComputer++
	case "cell phone":
		p.DeviceTypeCellPhone++
	case "smart tv device":
		p.DeviceTypeSmartTVDevice++
	case "connected audio device":
		p.DeviceTypeConnectedAudioDevice++
	case "built-in car application":
		p.DeviceTypeBuiltinCarApplication++
	default:
		p.DeviceTypeOther++
	}

	switch deviceOS := stream.DeviceOS; deviceOS {
	case "android":
		p.DeviceOSAndroid++
	case "blackberry":
		p.DeviceOSBlackBerry++
	case "browser":
		p.DeviceOSBrowser++
	case "ios":
		p.DeviceOSiOS++
	case "linux":
		p.DeviceOSLinux++
	case "mac":
		p.DeviceOSMac++
	case "windows":
		p.DeviceOSWindows++
	default:
		p.DeviceOSOther++
	}

	return p
}

func mutatePlaylistTrack(pt playlistTrackRecord, stream spotifyStreamRecord) playlistTrackRecord {
	pt.Streams++

	if stream.ShufflePlay == "y" {
		pt.ShufflePlay++
	}
	if stream.RepeatPlay == "y" {
		pt.RepeatPlay++
	}

	switch deviceType := stream.DeviceType; deviceType {
	case "desktop":
		pt.DeviceTypePersonalComputer++
	case "mobile":
		pt.DeviceTypeCellPhone++
	case "tablet":
		pt.DeviceTypeTablet++
	case "personal computer":
		pt.DeviceTypePersonalComputer++
	case "cell phone":
		pt.DeviceTypeCellPhone++
	case "smart tv device":
		pt.DeviceTypeSmartTVDevice++
	case "connected audio device":
		pt.DeviceTypeConnectedAudioDevice++
	case "built-in car application":
		pt.DeviceTypeBuiltinCarApplication++
	default:
		pt.DeviceTypeOther++
	}

	switch deviceOS := stream.DeviceOS; deviceOS {
	case "android":
		pt.DeviceOSAndroid++
	case "blackberry":
		pt.DeviceOSBlackBerry++
	case "browser":
		pt.DeviceOSBrowser++
	case "ios":
		pt.DeviceOSiOS++
	case "linux":
		pt.DeviceOSLinux++
	case "mac":
		pt.DeviceOSMac++
	case "windows":
		pt.DeviceOSWindows++
	default:
		pt.DeviceOSOther++
	}

	return pt
}

func processStreams(date string, licensor string, countryCode string, tracks map[string]spotifyTrack, users map[string]string) {
	input := &s3.GetObjectInput{
		Bucket: aws.String(s3SourceBucket),
		Key:    aws.String(makeSpotifySourceStreamsS3Key(date, licensor, countryCode)),
	}

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

	sMap := make(map[string]summaryRecord)
	tMap := make(map[string]trackRecord)
	pMap := make(map[string]playlistRecord)
	ptMap := make(map[string]playlistTrackRecord)

	scanner := bufio.NewScanner(result.Body)
	for scanner.Scan() {
		stream := unmarshalSpotifyStreamRecord([]byte(scanner.Text()))

		userCountryCode, ok := users[stream.UserID]
		if ok == true {
			track := tracks[stream.TrackID]

			summaryKey := userCountryCode
			s, ok := sMap[summaryKey]
			if ok == false {
				s = summaryRecord{
					ReportDate:        date,
					ReportCountryCode: countryCode,
					Date:              date,
					CountryCode:       userCountryCode,
				}
				sMap[summaryKey] = s
			}
			sMap[summaryKey] = mutateSummary(s, stream)

			if stream.Source == "others_playlist" && stream.SourceURI != "" {
				playlistKey := stream.SourceURI + "\xff" + userCountryCode
				p, ok := pMap[playlistKey]
				if ok == false {
					p = playlistRecord{
						ReportDate:        date,
						ReportCountryCode: countryCode,
						Date:              date,
						CountryCode:       userCountryCode,
						PlaylistURI:       stream.SourceURI,
					}
					pMap[playlistKey] = p
				}

				pMap[playlistKey] = mutatePlaylist(p, stream)
			}

			if track.ISRC != "" {
				trackKey := track.ISRC + "\xff" + userCountryCode

				t, ok := tMap[trackKey]
				if ok == false {
					t = trackRecord{
						ReportDate:        date,
						ReportCountryCode: countryCode,
						Date:              date,
						CountryCode:       userCountryCode,
						ISRC:              track.ISRC,
					}
					tMap[trackKey] = t
				}

				tMap[trackKey] = mutateTrack(t, stream)

				if stream.Source == "others_playlist" && stream.SourceURI != "" {
					playlistTrackKey := track.ISRC + "\xff" + stream.SourceURI + "\xff" + userCountryCode

					pt, ok := ptMap[playlistTrackKey]
					if ok == false {
						pt = playlistTrackRecord{
							ReportDate:        date,
							ReportCountryCode: countryCode,
							Date:              date,
							CountryCode:       userCountryCode,
							PlaylistURI:       stream.SourceURI,
							ISRC:              track.ISRC,
						}
						ptMap[playlistTrackKey] = pt
					}

					ptMap[playlistTrackKey] = mutatePlaylistTrack(pt, stream)
				}
			}
		}
	}
	if err := scanner.Err(); err != nil {
		log.Fatal(err)
	}

	tracksSession := session.Must(session.NewSession(&aws.Config{Region: aws.String("eu-west-1"), S3UseAccelerate: aws.Bool(true)}))
	tracksUploader := s3manager.NewUploader(tracksSession)
	tracksReader, tracksWriter := io.Pipe()
	playlistsSession := session.Must(session.NewSession(&aws.Config{Region: aws.String("eu-west-1"), S3UseAccelerate: aws.Bool(true)}))
	playlistsUploader := s3manager.NewUploader(playlistsSession)
	playlistsReader, playlistsWriter := io.Pipe()
	playlistTracksSession := session.Must(session.NewSession(&aws.Config{Region: aws.String("eu-west-1"), S3UseAccelerate: aws.Bool(true)}))
	playlistTracksUploader := s3manager.NewUploader(playlistTracksSession)
	playlistTracksReader, playlistTracksWriter := io.Pipe()
	summarySession := session.Must(session.NewSession(&aws.Config{Region: aws.String("eu-west-1"), S3UseAccelerate: aws.Bool(true)}))
	summaryUploader := s3manager.NewUploader(summarySession)
	summaryReader, summaryWriter := io.Pipe()

	go func() {
		gw := gzip.NewWriter(tracksWriter)
		for _, v := range tMap {
			row, err := json.Marshal(v)
			if err != nil {
				log.Fatal(err)
			}

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

			n1, err := gw.Write(r)
			if err != nil {
				log.Fatal(err)
			}
			_ = n1
		}
		gw.Close()
		tracksWriter.Close()
	}()

	go func() {
		gw := gzip.NewWriter(playlistsWriter)
		for _, v := range pMap {
			row, err := json.Marshal(v)
			if err != nil {
				log.Fatal(err)
			}

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

			n1, err := gw.Write(r)
			if err != nil {
				log.Fatal(err)
			}
			_ = n1
		}
		gw.Close()
		playlistsWriter.Close()
	}()

	go func() {
		gw := gzip.NewWriter(playlistTracksWriter)
		for _, v := range ptMap {
			row, err := json.Marshal(v)
			if err != nil {
				log.Fatal(err)
			}

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

			n1, err := gw.Write(r)
			if err != nil {
				log.Fatal(err)
			}
			_ = n1
		}
		gw.Close()
		playlistTracksWriter.Close()
	}()

	go func() {
		gw := gzip.NewWriter(summaryWriter)
		for _, v := range sMap {
			row, err := json.Marshal(v)
			if err != nil {
				log.Fatal(err)
			}

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

			n1, err := gw.Write(r)
			if err != nil {
				log.Fatal(err)
			}
			_ = n1
		}
		gw.Close()
		summaryWriter.Close()
	}()

	tracksResult, err := tracksUploader.Upload(&s3manager.UploadInput{
		Bucket:          aws.String(s3TargetBucket),
		Key:             aws.String(makeSpotifyTargetStreamsTracksS3Key(date, licensor, countryCode)),
		Body:            tracksReader,
		ContentType:     aws.String("application/x-ndjson"),
		ContentEncoding: aws.String("gzip"),
	})
	if err != nil {
		log.Fatal(err)
	}
	_ = tracksResult

	playlistsResult, err := playlistsUploader.Upload(&s3manager.UploadInput{
		Bucket:          aws.String(s3TargetBucket),
		Key:             aws.String(makeSpotifyTargetStreamsPlaylistsS3Key(date, licensor, countryCode)),
		Body:            playlistsReader,
		ContentType:     aws.String("application/x-ndjson"),
		ContentEncoding: aws.String("gzip"),
	})
	if err != nil {
		log.Fatal(err)
	}
	_ = playlistsResult

	playlistTracksResult, err := playlistTracksUploader.Upload(&s3manager.UploadInput{
		Bucket:          aws.String(s3TargetBucket),
		Key:             aws.String(makeSpotifyTargetStreamsPlaylistTracksS3Key(date, licensor, countryCode)),
		Body:            playlistTracksReader,
		ContentType:     aws.String("application/x-ndjson"),
		ContentEncoding: aws.String("gzip"),
	})
	if err != nil {
		log.Fatal(err)
	}
	_ = playlistTracksResult

	summaryResult, err := summaryUploader.Upload(&s3manager.UploadInput{
		Bucket:          aws.String(s3TargetBucket),
		Key:             aws.String(makeSpotifyTargetStreamsSummaryS3Key(date, licensor, countryCode)),
		Body:            summaryReader,
		ContentType:     aws.String("application/x-ndjson"),
		ContentEncoding: aws.String("gzip"),
	})
	if err != nil {
		log.Fatal(err)
	}
	_ = summaryResult
}

func runTask(date string, licensor string, countryCode string) {
	summaryExist := targetSpotifyStreamsSummaryFileExists(date, licensor, countryCode)
	if summaryExist == true {
		log.Println("already processed", date, licensor, countryCode)
		return
	}

	streamsExist := sourceSpotifyStreamsFileExists(date, licensor, countryCode)
	if streamsExist == false {
		log.Println("streams doesn't exist", date, licensor, countryCode)
		return
	}
	usersExist := sourceSpotifyUsersFileExists(date, licensor)
	if usersExist == false {
		log.Println("users doesn't exist", date, licensor)
		return
	}
	tracksExist := sourceSpotifyTracksFileExists(date, licensor)
	if tracksExist == false {
		log.Println("tracks doesn't exist", date, licensor)
		return
	}

	log.Println("fetching users", date, licensor)
	users := getSpotifyUsers(date, licensor)

	log.Println("fetching tracks", date, licensor)
	tracks := getSpotifyTracks(date, licensor)

	log.Println("starting", date, licensor, countryCode)
	processStreams(date, licensor, countryCode, tracks, users)
}

func main() {
	flag.Parse()

	startDate, err := time.Parse(time.RFC3339, inputStartDate+"T00:00:00Z")
	if err != nil {
		log.Fatal(err)
	}
	endDate, err := time.Parse(time.RFC3339, inputEndDate+"T00:00:00Z")
	if err != nil {
		log.Fatal(err)
	}

	dates := dateRange(startDate, endDate)
	sort.Sort(sort.Reverse(sort.StringSlice(dates)))

	for d := range dates {
		date := dates[d]
		for l := range spotifyLicensors {
			licensor := spotifyLicensors[l]
			if licensor == "sony" && date >= "2017-09-04" {
				for c := range spotifyV2CountryCodes {
					countryCode := spotifyV2CountryCodes[c]
					runTask(date, licensor, countryCode)
				}
			} else {
				for c := range spotifyV1CountryCodes {
					countryCode := spotifyV1CountryCodes[c]
					runTask(date, licensor, countryCode)
				}
			}
		}
	}
}