package spotifyconsumeranalytics

import (
	"bufio"
	"compress/gzip"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"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 spotifyStreamRecord struct {
	Date        string `json:"date"`
	UserID      string `json:"user_id"`
	TrackID     string `json:"track_id"`
	Source      string `json:"source"`
	SourceURI   string `json:"source_uri"`
	DeviceType  string `json:"device_type"`
	Shuffle     string `json:"shuffle"`
	ShufflePlay string `json:"shuffle_play"`
	RepeatPlay  string `json:"repeat_play"`
	DeviceOS    string `json:"os"`
}

type summaryRecord 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"`
	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"`
	FreeListeners                     int    `json:"free_listeners"`
	PaidListeners                     int    `json:"paid_listeners"`
	Listeners                         int    `json:"listeners"`
	FreeListenersSourceOthersPlaylist int    `json:"free_listeners_source_others_playlist"`
	PaidListenersSourceOthersPlaylist int    `json:"paid_listeners_source_others_playlist"`
	ListenersSourceOthersPlaylist     int    `json:"listeners_source_others_playlist"`
	RepeatPlay                        int    `json:"repeat_play"`
	ShufflePlay                       int    `json:"shuffle_play"`
	FreeSourceAlbum                   int    `json:"free_source_album"`
	FreeSourceArtist                  int    `json:"free_source_artist"`
	FreeSourceChart                   int    `json:"free_source_chart"`
	FreeSourceCollection              int    `json:"free_source_collection"`
	FreeSourceDailyMix                int    `json:"free_source_daily_mix"`
	FreeSourceDiscoverWeekly          int    `json:"free_source_discover_weekly"`
	FreeSourceOther                   int    `json:"free_source_other"`
	FreeSourceOthersPlaylist          int    `json:"free_source_others_playlist"`
	FreeSourcePlayQueue               int    `json:"free_source_play_queue"`
	FreeSourceRadio                   int    `json:"free_source_radio"`
	FreeSourceReleaseRadar            int    `json:"free_source_release_radar"`
	FreeSourceSearch                  int    `json:"free_source_search"`
	FreeSourceUnknown                 int    `json:"free_source_unknown"`
	PaidSourceAlbum                   int    `json:"paid_source_album"`
	PaidSourceArtist                  int    `json:"paid_source_artist"`
	PaidSourceChart                   int    `json:"paid_source_chart"`
	PaidSourceCollection              int    `json:"paid_source_collection"`
	PaidSourceDailyMix                int    `json:"paid_source_daily_mix"`
	PaidSourceDiscoverWeekly          int    `json:"paid_source_discover_weekly"`
	PaidSourceOther                   int    `json:"paid_source_other"`
	PaidSourceOthersPlaylist          int    `json:"paid_source_others_playlist"`
	PaidSourcePlayQueue               int    `json:"paid_source_play_queue"`
	PaidSourceRadio                   int    `json:"paid_source_radio"`
	PaidSourceReleaseRadar            int    `json:"paid_source_release_radar"`
	PaidSourceSearch                  int    `json:"paid_source_search"`
	PaidSourceUnknown                 int    `json:"paid_source_unknown"`
	SourceAlbum                       int    `json:"source_album"`
	SourceArtist                      int    `json:"source_artist"`
	SourceChart                       int    `json:"source_chart"`
	SourceCollection                  int    `json:"source_collection"`
	SourceDailyMix                    int    `json:"source_daily_mix"`
	SourceDiscoverWeekly              int    `json:"source_discover_weekly"`
	SourceOther                       int    `json:"source_other"`
	SourceOthersPlaylist              int    `json:"source_others_playlist"`
	SourcePlayQueue                   int    `json:"source_play_queue"`
	SourceRadio                       int    `json:"source_radio"`
	SourceReleaseRadar                int    `json:"source_release_radar"`
	SourceSearch                      int    `json:"source_search"`
	SourceUnknown                     int    `json:"source_unknown"`
	FreeStreams                       int    `json:"free_streams"`
	PaidStreams                       int    `json:"paid_streams"`
	Streams                           int    `json:"streams"`
}

type trackRecord 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"`
	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"`
	FreeListeners                   int    `json:"free_listeners"`
	PaidListeners                   int    `json:"paid_listeners"`
	Listeners                       int    `json:"listeners"`
	RepeatPlay                      int    `json:"repeat_play"`
	ShufflePlay                     int    `json:"shuffle_play"`
	FreeSourceAlbum                 int    `json:"free_source_album"`
	FreeSourceArtist                int    `json:"free_source_artist"`
	FreeSourceChart                 int    `json:"free_source_chart"`
	FreeSourceCollection            int    `json:"free_source_collection"`
	FreeSourceDailyMix              int    `json:"free_source_daily_mix"`
	FreeSourceDiscoverWeekly        int    `json:"free_source_discover_weekly"`
	FreeSourceOther                 int    `json:"free_source_other"`
	FreeSourceOthersPlaylist        int    `json:"free_source_others_playlist"`
	FreeSourcePlayQueue             int    `json:"free_source_play_queue"`
	FreeSourceRadio                 int    `json:"free_source_radio"`
	FreeSourceReleaseRadar          int    `json:"free_source_release_radar"`
	FreeSourceSearch                int    `json:"free_source_search"`
	FreeSourceUnknown               int    `json:"free_source_unknown"`
	PaidSourceAlbum                 int    `json:"paid_source_album"`
	PaidSourceArtist                int    `json:"paid_source_artist"`
	PaidSourceChart                 int    `json:"paid_source_chart"`
	PaidSourceCollection            int    `json:"paid_source_collection"`
	PaidSourceDailyMix              int    `json:"paid_source_daily_mix"`
	PaidSourceDiscoverWeekly        int    `json:"paid_source_discover_weekly"`
	PaidSourceOther                 int    `json:"paid_source_other"`
	PaidSourceOthersPlaylist        int    `json:"paid_source_others_playlist"`
	PaidSourcePlayQueue             int    `json:"paid_source_play_queue"`
	PaidSourceRadio                 int    `json:"paid_source_radio"`
	PaidSourceReleaseRadar          int    `json:"paid_source_release_radar"`
	PaidSourceSearch                int    `json:"paid_source_search"`
	PaidSourceUnknown               int    `json:"paid_source_unknown"`
	SourceAlbum                     int    `json:"source_album"`
	SourceArtist                    int    `json:"source_artist"`
	SourceChart                     int    `json:"source_chart"`
	SourceCollection                int    `json:"source_collection"`
	SourceDailyMix                  int    `json:"source_daily_mix"`
	SourceDiscoverWeekly            int    `json:"source_discover_weekly"`
	SourceOther                     int    `json:"source_other"`
	SourceOthersPlaylist            int    `json:"source_others_playlist"`
	SourcePlayQueue                 int    `json:"source_play_queue"`
	SourceRadio                     int    `json:"source_radio"`
	SourceReleaseRadar              int    `json:"source_release_radar"`
	SourceSearch                    int    `json:"source_search"`
	SourceUnknown                   int    `json:"source_unknown"`
	FreeStreams                     int    `json:"free_streams"`
	PaidStreams                     int    `json:"paid_streams"`
	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"`
	Licensor                        string `json:"licensor"`
	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"`
	FreeListeners                   int    `json:"free_listeners"`
	PaidListeners                   int    `json:"paid_listeners"`
	Listeners                       int    `json:"listeners"`
	RepeatPlay                      int    `json:"repeat_play"`
	ShufflePlay                     int    `json:"shuffle_play"`
	FreeStreams                     int    `json:"free_streams"`
	PaidStreams                     int    `json:"paid_streams"`
	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"`
	Licensor                        string `json:"licensor"`
	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"`
	FreeListeners                   int    `json:"free_listeners"`
	PaidListeners                   int    `json:"paid_listeners"`
	Listeners                       int    `json:"listeners"`
	RepeatPlay                      int    `json:"repeat_play"`
	ShufflePlay                     int    `json:"shuffle_play"`
	FreeStreams                     int    `json:"free_streams"`
	PaidStreams                     int    `json:"paid_streams"`
	Streams                         int    `json:"streams"`
}

func unmarshalSpotifyStreamRecord(data []byte) (spotifyStreamRecord, error) {
	var stream spotifyStreamRecord
	err := json.Unmarshal(data, &stream)
	if err != nil {
		return stream, fmt.Errorf("unmarshalSpotifyStreamRecord: %s", err)
	}
	stream.DeviceOS = strings.ToLower(stream.DeviceOS)
	if stream.Shuffle != "" && stream.ShufflePlay == "" {
		stream.ShufflePlay = stream.Shuffle
	}
	stream.ShufflePlay = strings.ToLower(stream.ShufflePlay)
	stream.RepeatPlay = strings.ToLower(stream.RepeatPlay)
	return stream, nil
}

func unmarshalSpotifyStreamRecordWithCollectionFix(data []byte) (spotifyStreamRecord, error) {
	var stream spotifyStreamRecord
	err := json.Unmarshal(data, &stream)
	if err != nil {
		return stream, fmt.Errorf("unmarshalSpotifyStreamRecordWithCollectionFix: %s", err)
	}
	stream.DeviceOS = strings.ToLower(stream.DeviceOS)
	if stream.Shuffle != "" && stream.ShufflePlay == "" {
		stream.ShufflePlay = stream.Shuffle
	}
	stream.ShufflePlay = strings.ToLower(stream.ShufflePlay)
	stream.RepeatPlay = strings.ToLower(stream.RepeatPlay)
	if stream.Source == "others_playlist" && stream.SourceURI == "" {
		stream.Source = "collection"
	}
	return stream, nil
}

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

	if user.Access == "free" {
		s.FreeStreams++
	} else {
		s.PaidStreams++
	}

	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 "play_queue":
		s.SourcePlayQueue++
	case "search":
		s.SourceSearch++
	case "":
		s.SourceUnknown++
	}

	if user.Access == "free" {
		switch source := stream.Source; source {
		case "album":
			s.FreeSourceAlbum++
		case "artist":
			s.FreeSourceArtist++
		case "chart":
			s.FreeSourceChart++
		case "collection":
			s.FreeSourceCollection++
		case "other":
			s.FreeSourceOther++
		case "play_queue":
			s.FreeSourcePlayQueue++
		case "search":
			s.FreeSourceSearch++
		case "":
			s.FreeSourceUnknown++
		}
	} else {
		switch source := stream.Source; source {
		case "album":
			s.PaidSourceAlbum++
		case "artist":
			s.PaidSourceArtist++
		case "chart":
			s.PaidSourceChart++
		case "collection":
			s.PaidSourceCollection++
		case "other":
			s.PaidSourceOther++
		case "play_queue":
			s.PaidSourcePlayQueue++
		case "search":
			s.PaidSourceSearch++
		case "":
			s.PaidSourceUnknown++
		}
	}

	if stream.Source == "others_playlist" {
		if stream.SourceURI == "discoverWeekly" {
			s.SourceDiscoverWeekly++
		} else if stream.SourceURI == "releaseRadar" {
			s.SourceReleaseRadar++
		} else {
			s.SourceOthersPlaylist++
		}

		if user.Access == "free" {
			if stream.SourceURI == "discoverWeekly" {
				s.FreeSourceDiscoverWeekly++
			} else if stream.SourceURI == "releaseRadar" {
				s.FreeSourceReleaseRadar++
			} else {
				s.FreeSourceOthersPlaylist++
			}
		} else {
			if stream.SourceURI == "discoverWeekly" {
				s.PaidSourceDiscoverWeekly++
			} else if stream.SourceURI == "releaseRadar" {
				s.PaidSourceReleaseRadar++
			} else {
				s.PaidSourceOthersPlaylist++
			}
		}
	}

	if stream.Source == "radio" {
		if stream.SourceURI == "dailyMix" {
			s.SourceDailyMix++
		} else {
			s.SourceRadio++
		}

		if user.Access == "free" {
			if stream.SourceURI == "dailyMix" {
				s.FreeSourceDailyMix++
			} else {
				s.FreeSourceRadio++
			}
		} else {
			if stream.SourceURI == "dailyMix" {
				s.PaidSourceDailyMix++
			} else {
				s.PaidSourceRadio++
			}
		}
	}

	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 mutateTrack(t trackRecord, stream spotifyStreamRecord, user spotifyUser) trackRecord {
	t.Streams++

	if user.Access == "free" {
		t.FreeStreams++
	} else {
		t.PaidStreams++
	}

	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 "play_queue":
		t.SourcePlayQueue++
	case "search":
		t.SourceSearch++
	case "":
		t.SourceUnknown++
	}

	if user.Access == "free" {
		switch source := stream.Source; source {
		case "album":
			t.FreeSourceAlbum++
		case "artist":
			t.FreeSourceArtist++
		case "chart":
			t.FreeSourceChart++
		case "collection":
			t.FreeSourceCollection++
		case "other":
			t.FreeSourceOther++
		case "play_queue":
			t.FreeSourcePlayQueue++
		case "search":
			t.FreeSourceSearch++
		case "":
			t.FreeSourceUnknown++
		}
	} else {
		switch source := stream.Source; source {
		case "album":
			t.PaidSourceAlbum++
		case "artist":
			t.PaidSourceArtist++
		case "chart":
			t.PaidSourceChart++
		case "collection":
			t.PaidSourceCollection++
		case "other":
			t.PaidSourceOther++
		case "play_queue":
			t.PaidSourcePlayQueue++
		case "search":
			t.PaidSourceSearch++
		case "":
			t.PaidSourceUnknown++
		}
	}

	if stream.Source == "others_playlist" {
		if stream.SourceURI == "discoverWeekly" {
			t.SourceDiscoverWeekly++
		} else if stream.SourceURI == "releaseRadar" {
			t.SourceReleaseRadar++
		} else {
			t.SourceOthersPlaylist++
		}

		if user.Access == "free" {
			if stream.SourceURI == "discoverWeekly" {
				t.FreeSourceDiscoverWeekly++
			} else if stream.SourceURI == "releaseRadar" {
				t.FreeSourceReleaseRadar++
			} else {
				t.FreeSourceOthersPlaylist++
			}
		} else {
			if stream.SourceURI == "discoverWeekly" {
				t.PaidSourceDiscoverWeekly++
			} else if stream.SourceURI == "releaseRadar" {
				t.PaidSourceReleaseRadar++
			} else {
				t.PaidSourceOthersPlaylist++
			}
		}
	}

	if stream.Source == "radio" {
		if stream.SourceURI == "dailyMix" {
			t.SourceDailyMix++
		} else {
			t.SourceRadio++
		}

		if user.Access == "free" {
			if stream.SourceURI == "dailyMix" {
				t.FreeSourceDailyMix++
			} else {
				t.FreeSourceRadio++
			}
		} else {
			if stream.SourceURI == "dailyMix" {
				t.PaidSourceDailyMix++
			} else {
				t.PaidSourceRadio++
			}
		}
	}

	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 mutatePlaylist(p playlistRecord, stream spotifyStreamRecord, user spotifyUser) playlistRecord {
	p.Streams++

	if user.Access == "free" {
		p.FreeStreams++
	} else {
		p.PaidStreams++
	}

	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, user spotifyUser) playlistTrackRecord {
	pt.Streams++

	if user.Access == "free" {
		pt.FreeStreams++
	} else {
		pt.PaidStreams++
	}

	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(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket, date, licensor, reportCountryCode string, tracks map[string]spotifyTrack, users map[string]spotifyUser) {
	input := &s3.GetObjectInput{
		Bucket: aws.String(s3SourceBucket),
		Key:    aws.String(makeSpotifySourceStreamsS3Key(date, licensor, reportCountryCode)),
	}

	result, err := s3SourceService.GetObject(input)
	if err != nil {
		log.Fatalf("processStreams: %s", err)
	}

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

	listenerSummaryMap := make(map[string]map[string]int)
	freeListenerSummaryMap := make(map[string]map[string]int)
	paidListenerSummaryMap := make(map[string]map[string]int)
	listenerSummaryPlaylistMap := make(map[string]map[string]int)
	freeListenerSummaryPlaylistMap := make(map[string]map[string]int)
	paidListenerSummaryPlaylistMap := make(map[string]map[string]int)
	listenerTrackMap := make(map[string]map[string]int)
	freeListenerTrackMap := make(map[string]map[string]int)
	paidListenerTrackMap := make(map[string]map[string]int)
	listenerPlaylistMap := make(map[string]map[string]int)
	freeListenerPlaylistMap := make(map[string]map[string]int)
	paidListenerPlaylistMap := make(map[string]map[string]int)
	listenerPlaylistTrackMap := make(map[string]map[string]int)
	freeListenerPlaylistTrackMap := make(map[string]map[string]int)
	paidListenerPlaylistTrackMap := make(map[string]map[string]int)

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

	scanner := bufio.NewScanner(reader)
	for scanner.Scan() {
		var stream spotifyStreamRecord
		var err error
		if date >= "2018-05-25" && date <= "2018-09-04" {
			stream, err = unmarshalSpotifyStreamRecordWithCollectionFix([]byte(scanner.Text()))
		} else {
			stream, err = unmarshalSpotifyStreamRecord([]byte(scanner.Text()))
		}
		if err != nil {
			log.Fatalf("processStreams: %s", err)
		}

		user, ok := users[stream.UserID]
		if ok == true {
			track := tracks[stream.TrackID]
			userCountryCode := user.CountryCode
			if userCountryCode == "" {
				userCountryCode = reportCountryCode
			}

			summaryKey := userCountryCode

			s, ok := sMap[summaryKey]
			if ok == false {
				s = summaryRecord{
					ReportDate:        date,
					ReportCountryCode: reportCountryCode,
					Date:              date,
					CountryCode:       userCountryCode,
					Licensor:          licensor,
				}
				sMap[summaryKey] = s
			}

			ls, ok2 := listenerSummaryMap[summaryKey]
			if ok2 == false {
				m := make(map[string]int)
				m[stream.UserID] = 1
				listenerSummaryMap[summaryKey] = m
			} else {
				ls[stream.UserID] = 1
				listenerSummaryMap[summaryKey] = ls
			}

			if user.Access == "free" {
				fls, ok2 := freeListenerSummaryMap[summaryKey]
				if ok2 == false {
					m := make(map[string]int)
					m[stream.UserID] = 1
					freeListenerSummaryMap[summaryKey] = m
				} else {
					fls[stream.UserID] = 1
					freeListenerSummaryMap[summaryKey] = fls
				}
			} else {
				pls, ok2 := paidListenerSummaryMap[summaryKey]
				if ok2 == false {
					m := make(map[string]int)
					m[stream.UserID] = 1
					paidListenerSummaryMap[summaryKey] = m
				} else {
					pls[stream.UserID] = 1
					paidListenerSummaryMap[summaryKey] = pls
				}
			}

			if stream.Source == "others_playlist" {
				lsp, ok2 := listenerSummaryPlaylistMap[summaryKey]
				if ok2 == false {
					m := make(map[string]int)
					m[stream.UserID] = 1
					listenerSummaryPlaylistMap[summaryKey] = m
				} else {
					lsp[stream.UserID] = 1
					listenerSummaryPlaylistMap[summaryKey] = lsp
				}

				if user.Access == "free" {
					flsp, ok2 := freeListenerSummaryPlaylistMap[summaryKey]
					if ok2 == false {
						m := make(map[string]int)
						m[stream.UserID] = 1
						freeListenerSummaryPlaylistMap[summaryKey] = m
					} else {
						flsp[stream.UserID] = 1
						freeListenerSummaryPlaylistMap[summaryKey] = flsp
					}
				} else {
					plsp, ok2 := paidListenerSummaryPlaylistMap[summaryKey]
					if ok2 == false {
						m := make(map[string]int)
						m[stream.UserID] = 1
						paidListenerSummaryPlaylistMap[summaryKey] = m
					} else {
						plsp[stream.UserID] = 1
						paidListenerSummaryPlaylistMap[summaryKey] = plsp
					}
				}
			}

			sMap[summaryKey] = mutateSummary(s, stream, user)

			if strings.HasPrefix(stream.SourceURI, "spotify:") {
				playlistKey := stream.SourceURI + "\xff" + userCountryCode
				p, ok := pMap[playlistKey]
				if ok == false {
					p = playlistRecord{
						ReportDate:        date,
						ReportCountryCode: reportCountryCode,
						Date:              date,
						CountryCode:       userCountryCode,
						Licensor:          licensor,
						PlaylistURI:       stream.SourceURI,
					}
					pMap[playlistKey] = p
				}

				lp, ok2 := listenerPlaylistMap[playlistKey]
				if ok2 == false {
					m := make(map[string]int)
					m[stream.UserID] = 1
					listenerPlaylistMap[playlistKey] = m
				} else {
					lp[stream.UserID] = 1
					listenerPlaylistMap[playlistKey] = lp
				}

				if user.Access == "free" {
					flp, ok2 := freeListenerPlaylistMap[playlistKey]
					if ok2 == false {
						m := make(map[string]int)
						m[stream.UserID] = 1
						freeListenerPlaylistMap[playlistKey] = m
					} else {
						flp[stream.UserID] = 1
						freeListenerPlaylistMap[playlistKey] = flp
					}
				} else {
					plp, ok2 := paidListenerPlaylistMap[playlistKey]
					if ok2 == false {
						m := make(map[string]int)
						m[stream.UserID] = 1
						paidListenerPlaylistMap[playlistKey] = m
					} else {
						plp[stream.UserID] = 1
						paidListenerPlaylistMap[playlistKey] = plp
					}
				}

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

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

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

				lt, ok2 := listenerTrackMap[trackKey]
				if ok2 == false {
					m := make(map[string]int)
					m[stream.UserID] = 1
					listenerTrackMap[trackKey] = m
				} else {
					lt[stream.UserID] = 1
					listenerTrackMap[trackKey] = lt
				}

				if user.Access == "free" {
					flt, ok2 := freeListenerTrackMap[trackKey]
					if ok2 == false {
						m := make(map[string]int)
						m[stream.UserID] = 1
						freeListenerTrackMap[trackKey] = m
					} else {
						flt[stream.UserID] = 1
						freeListenerTrackMap[trackKey] = flt
					}
				} else {
					plt, ok2 := paidListenerTrackMap[trackKey]
					if ok2 == false {
						m := make(map[string]int)
						m[stream.UserID] = 1
						paidListenerTrackMap[trackKey] = m
					} else {
						plt[stream.UserID] = 1
						paidListenerTrackMap[trackKey] = plt
					}
				}

				tMap[trackKey] = mutateTrack(t, stream, user)

				if stream.Source == "others_playlist" && strings.HasPrefix(stream.SourceURI, "spotify:") {
					playlistTrackKey := track.ISRC + "\xff" + stream.SourceURI + "\xff" + userCountryCode

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

					lpt, ok2 := listenerPlaylistTrackMap[playlistTrackKey]
					if ok2 == false {
						m := make(map[string]int)
						m[stream.UserID] = 1
						listenerPlaylistTrackMap[playlistTrackKey] = m
					} else {
						lpt[stream.UserID] = 1
						listenerPlaylistTrackMap[playlistTrackKey] = lpt
					}

					if user.Access == "free" {
						flpt, ok2 := freeListenerPlaylistTrackMap[playlistTrackKey]
						if ok2 == false {
							m := make(map[string]int)
							m[stream.UserID] = 1
							freeListenerPlaylistTrackMap[playlistTrackKey] = m
						} else {
							flpt[stream.UserID] = 1
							freeListenerPlaylistTrackMap[playlistTrackKey] = flpt
						}
					} else {
						plpt, ok2 := paidListenerPlaylistTrackMap[playlistTrackKey]
						if ok2 == false {
							m := make(map[string]int)
							m[stream.UserID] = 1
							paidListenerPlaylistTrackMap[playlistTrackKey] = m
						} else {
							plpt[stream.UserID] = 1
							paidListenerPlaylistTrackMap[playlistTrackKey] = plpt
						}
					}

					ptMap[playlistTrackKey] = mutatePlaylistTrack(pt, stream, user)
				}
			}
		}
	}
	if err := scanner.Err(); err != nil {
		log.Fatalf("processStreams: %s", err)
	}

	for k, v := range sMap {
		v.FreeListeners = len(freeListenerSummaryMap[k])
		v.PaidListeners = len(paidListenerSummaryMap[k])
		v.Listeners = len(listenerSummaryMap[k])
		v.FreeListenersSourceOthersPlaylist = len(freeListenerSummaryPlaylistMap[k])
		v.PaidListenersSourceOthersPlaylist = len(paidListenerSummaryPlaylistMap[k])
		v.ListenersSourceOthersPlaylist = len(listenerSummaryPlaylistMap[k])
		sMap[k] = v
	}

	for k, v := range tMap {
		v.FreeListeners = len(freeListenerTrackMap[k])
		v.PaidListeners = len(paidListenerTrackMap[k])
		v.Listeners = len(listenerTrackMap[k])
		tMap[k] = v
	}

	for k, v := range pMap {
		v.FreeListeners = len(paidListenerPlaylistMap[k])
		v.PaidListeners = len(freeListenerPlaylistMap[k])
		v.Listeners = len(listenerPlaylistMap[k])
		pMap[k] = v
	}

	for k, v := range ptMap {
		v.FreeListeners = len(freeListenerPlaylistTrackMap[k])
		v.PaidListeners = len(paidListenerPlaylistTrackMap[k])
		v.Listeners = len(listenerPlaylistTrackMap[k])
		ptMap[k] = v
	}

	tracksSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)}))
	tracksUploader := s3manager.NewUploader(tracksSession)
	tracksReader, tracksWriter := io.Pipe()
	playlistsSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)}))
	playlistsUploader := s3manager.NewUploader(playlistsSession)
	playlistsReader, playlistsWriter := io.Pipe()
	playlistTracksSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)}))
	playlistTracksUploader := s3manager.NewUploader(playlistTracksSession)
	playlistTracksReader, playlistTracksWriter := io.Pipe()
	summarySession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)}))
	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.Fatalf("processStreams, tracksWriter: %s", err)
			}

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

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

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

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

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

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

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

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

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

			r := append(row, []byte("\n")...)
			if _, err = gw.Write(r); err != nil {
				log.Fatalf("processStreams, summaryWriter: %s", err)
			}
		}
		if err = gw.Close(); err != nil {
			log.Fatalf("processStreams, summaryWriter: %s", err)
		}
		if err = summaryWriter.Close(); err != nil {
			log.Fatalf("processStreams, summaryWriter: %s", err)
		}
	}()

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

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

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

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

func summaryExists(s3Service *s3.S3, s3TargetBucket, key string) bool {
	input := &s3.HeadObjectInput{
		Bucket: aws.String(s3TargetBucket),
		Key:    aws.String(key),
	}

	result, err := s3Service.HeadObject(input)
	if err != nil {
		return false
	}

	if result.LastModified.Unix() < 1526553733 {
		return false
	}

	return true
}

func runTask(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket, date, licensor, reportCountryCode string) {
	summarySourceKey := makeSpotifyTargetStreamsSummaryS3Key(date, licensor, reportCountryCode)
	if summaryExists(s3TargetService, s3TargetBucket, summarySourceKey) == true {
		log.Printf("runTask, already processed: %s %s %s %s", summarySourceKey, date, licensor, reportCountryCode)
		return
	}

	if ok, err := sourceSpotifyStreamsFileExists(s3SourceService, s3SourceBucket, date, licensor, reportCountryCode); err != nil {
		log.Printf("runTask, sourceSpotifyStreamsFileExists, error: %s", err)
		return
	} else if !ok {
		log.Printf("runTask, streams doesn't exist: %s %s %s", date, licensor, reportCountryCode)
		return
	}
	if ok, err := sourceSpotifyUsersFileExists(s3SourceService, s3SourceBucket, date, licensor); err != nil {
		log.Printf("runTask, sourceSpotifyUsersFileExists, error: %s", err)
		return
	} else if !ok {
		log.Printf("runTask, users doesn't exist: %s %s", date, licensor)
		return
	}
	if ok, err := sourceSpotifyTracksFileExists(s3SourceService, s3SourceBucket, date, licensor); err != nil {
		log.Printf("runTask, sourceSpotifyTracksFileExists, error: %s", err)
		return
	} else if !ok {
		log.Printf("runTask, tracks doesn't exist: %s %s", date, licensor)
		return
	}

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

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

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

// RunStreams aggregates streams and uploads to S3
func RunStreams(s3SourceService *s3.S3, s3TargetService *s3.S3, startDate time.Time, endDate time.Time, licensors, countryCodes []string, s3SourceBucket, s3TargetBucket string) {
	dates := daterange.DateRange(startDate, endDate)

	if len(countryCodes) > 1 {
		for _, date := range dates {
			for _, licensor := range licensors {
				for _, reportCountryCode := range countryCodes {
					runTask(s3SourceService, s3TargetService, s3SourceBucket, s3TargetBucket, date, licensor, reportCountryCode)
				}
			}
		}
	} else {
		for _, date := range dates {
			for _, licensor := range licensors {
				if date >= "2018-01-22" {
					for _, reportCountryCode := range V2CountryCodes {
						runTask(s3SourceService, s3TargetService, s3SourceBucket, s3TargetBucket, date, licensor, reportCountryCode)
					}
				} else {
					for _, reportCountryCode := range V1CountryCodes {
						runTask(s3SourceService, s3TargetService, s3SourceBucket, s3TargetBucket, date, licensor, reportCountryCode)
					}
				}
			}
		}
	}
}