package appleconsumeranalytics

import (
	"bufio"
	"compress/gzip"
	"encoding/json"
	"io"
	"log"
	"strconv"
	"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 reportRecord struct {
	Datestamp       string
	IngestDatestamp string
	AppleIdentifier string
	StorefrontName  string
	MembershipType  string
	MembershipMode  string
	Gender          string
	AgeBand         string
	Listeners       int
	Streams         int
}

func parseReportRow(headers []string, row []string) reportRecord {
	itemMap := make(map[string]string)
	for i, k := range headers {
		key := strings.ToLower(strings.TrimSpace(k))
		value := strings.TrimSpace(row[i])
		itemMap[key] = value
	}
	reportRecord := reportRecord{
		Datestamp:       itemMap["datestamp"],
		IngestDatestamp: itemMap["ingest datestamp"],
		AppleIdentifier: itemMap["apple identifier"],
		StorefrontName:  itemMap["storefront name"],
		MembershipType:  itemMap["membership type"],
		MembershipMode:  itemMap["membership mode"],
		Gender:          itemMap["gender"],
		AgeBand:         itemMap["age band"],
	}

	listeners, err := strconv.Atoi(itemMap["listeners"])
	if err != nil {
		log.Fatal(err)
	}
	reportRecord.Listeners = listeners

	streams, err := strconv.Atoi(itemMap["streams"])
	if err != nil {
		log.Fatal(err)
	}
	reportRecord.Streams = streams

	return reportRecord
}

type contentDemographicsRecord struct {
	ReportDate                       string `json:"report_date"`
	ReportVendorID                   string `json:"report_vendor_id"`
	Date                             string `json:"date"`
	CountryCode                      string `json:"country_code"`
	Licensor                         string `json:"licensor"`
	ISRC                             string `json:"isrc"`
	ListenersGenderFemaleAge0_17     int    `json:"listeners_gender_female_age_0_17"`
	ListenersGenderMaleAge0_17       int    `json:"listeners_gender_male_age_0_17"`
	ListenersGenderUnknownAge0_17    int    `json:"listeners_gender_unknown_age_0_17"`
	ListenersGenderFemaleAge18_24    int    `json:"listeners_gender_female_age_18_24"`
	ListenersGenderMaleAge18_24      int    `json:"listeners_gender_male_age_18_24"`
	ListenersGenderUnknownAge18_24   int    `json:"listeners_gender_unknown_age_18_24"`
	ListenersGenderFemaleAge25_34    int    `json:"listeners_gender_female_age_25_34"`
	ListenersGenderMaleAge25_34      int    `json:"listeners_gender_male_age_25_34"`
	ListenersGenderUnknownAge25_34   int    `json:"listeners_gender_unknown_age_25_34"`
	ListenersGenderFemaleAge35_44    int    `json:"listeners_gender_female_age_35_44"`
	ListenersGenderMaleAge35_44      int    `json:"listeners_gender_male_age_35_44"`
	ListenersGenderUnknownAge35_44   int    `json:"listeners_gender_unknown_age_35_44"`
	ListenersGenderFemaleAge45_54    int    `json:"listeners_gender_female_age_45_54"`
	ListenersGenderMaleAge45_54      int    `json:"listeners_gender_male_age_45_54"`
	ListenersGenderUnknownAge45_54   int    `json:"listeners_gender_unknown_age_45_54"`
	ListenersGenderFemaleAge55_64    int    `json:"listeners_gender_female_age_55_64"`
	ListenersGenderMaleAge55_64      int    `json:"listeners_gender_male_age_55_64"`
	ListenersGenderUnknownAge55_64   int    `json:"listeners_gender_unknown_age_55_64"`
	ListenersGenderFemaleAge65Plus   int    `json:"listeners_gender_female_age_65_plus"`
	ListenersGenderMaleAge65Plus     int    `json:"listeners_gender_male_age_65_plus"`
	ListenersGenderUnknownAge65Plus  int    `json:"listeners_gender_unknown_age_65_plus"`
	ListenersGenderFemaleAgeUnknown  int    `json:"listeners_gender_female_age_unknown"`
	ListenersGenderMaleAgeUnknown    int    `json:"listeners_gender_male_age_unknown"`
	ListenersGenderUnknownAgeUnknown int    `json:"listeners_gender_unknown_age_unknown"`
	StreamsGenderFemaleAge0_17       int    `json:"streams_gender_female_age_0_17"`
	StreamsGenderMaleAge0_17         int    `json:"streams_gender_male_age_0_17"`
	StreamsGenderUnknownAge0_17      int    `json:"streams_gender_unknown_age_0_17"`
	StreamsGenderFemaleAge18_24      int    `json:"streams_gender_female_age_18_24"`
	StreamsGenderMaleAge18_24        int    `json:"streams_gender_male_age_18_24"`
	StreamsGenderUnknownAge18_24     int    `json:"streams_gender_unknown_age_18_24"`
	StreamsGenderFemaleAge25_34      int    `json:"streams_gender_female_age_25_34"`
	StreamsGenderMaleAge25_34        int    `json:"streams_gender_male_age_25_34"`
	StreamsGenderUnknownAge25_34     int    `json:"streams_gender_unknown_age_25_34"`
	StreamsGenderFemaleAge35_44      int    `json:"streams_gender_female_age_35_44"`
	StreamsGenderMaleAge35_44        int    `json:"streams_gender_male_age_35_44"`
	StreamsGenderUnknownAge35_44     int    `json:"streams_gender_unknown_age_35_44"`
	StreamsGenderFemaleAge45_54      int    `json:"streams_gender_female_age_45_54"`
	StreamsGenderMaleAge45_54        int    `json:"streams_gender_male_age_45_54"`
	StreamsGenderUnknownAge45_54     int    `json:"streams_gender_unknown_age_45_54"`
	StreamsGenderFemaleAge55_64      int    `json:"streams_gender_female_age_55_64"`
	StreamsGenderMaleAge55_64        int    `json:"streams_gender_male_age_55_64"`
	StreamsGenderUnknownAge55_64     int    `json:"streams_gender_unknown_age_55_64"`
	StreamsGenderFemaleAge65Plus     int    `json:"streams_gender_female_age_65_plus"`
	StreamsGenderMaleAge65Plus       int    `json:"streams_gender_male_age_65_plus"`
	StreamsGenderUnknownAge65Plus    int    `json:"streams_gender_unknown_age_65_plus"`
	StreamsGenderFemaleAgeUnknown    int    `json:"streams_gender_female_age_unknown"`
	StreamsGenderMaleAgeUnknown      int    `json:"streams_gender_male_age_unknown"`
	StreamsGenderUnknownAgeUnknown   int    `json:"streams_gender_unknown_age_unknown"`
	Listeners                        int    `json:"listeners"`
	Streams                          int    `json:"streams"`
}

func mutateContentDemographics(c contentDemographicsRecord, reportRecord reportRecord) contentDemographicsRecord {
	c.Listeners = c.Listeners + reportRecord.Listeners
	c.Streams = c.Streams + reportRecord.Streams

	if reportRecord.Gender == "1" {
		switch reportRecord.AgeBand {
		case "<18":
			c.ListenersGenderMaleAge0_17 = c.ListenersGenderMaleAge0_17 + reportRecord.Listeners
			c.StreamsGenderMaleAge0_17 = c.StreamsGenderMaleAge0_17 + reportRecord.Streams
		case "18-24":
			c.ListenersGenderMaleAge18_24 = c.ListenersGenderMaleAge18_24 + reportRecord.Listeners
			c.StreamsGenderMaleAge18_24 = c.StreamsGenderMaleAge18_24 + reportRecord.Streams
		case "25-34":
			c.ListenersGenderMaleAge25_34 = c.ListenersGenderMaleAge25_34 + reportRecord.Listeners
			c.StreamsGenderMaleAge25_34 = c.StreamsGenderMaleAge25_34 + reportRecord.Streams
		case "35-44":
			c.ListenersGenderMaleAge35_44 = c.ListenersGenderMaleAge35_44 + reportRecord.Listeners
			c.StreamsGenderMaleAge35_44 = c.StreamsGenderMaleAge35_44 + reportRecord.Streams
		case "45-54":
			c.ListenersGenderMaleAge45_54 = c.ListenersGenderMaleAge45_54 + reportRecord.Listeners
			c.StreamsGenderMaleAge45_54 = c.StreamsGenderMaleAge45_54 + reportRecord.Streams
		case "55-64":
			c.ListenersGenderMaleAge55_64 = c.ListenersGenderMaleAge55_64 + reportRecord.Listeners
			c.StreamsGenderMaleAge55_64 = c.StreamsGenderMaleAge55_64 + reportRecord.Streams
		case "65+":
			c.ListenersGenderMaleAge65Plus = c.ListenersGenderMaleAge65Plus + reportRecord.Listeners
			c.StreamsGenderMaleAge65Plus = c.StreamsGenderMaleAge65Plus + reportRecord.Streams
		default:
			c.ListenersGenderMaleAgeUnknown = c.ListenersGenderMaleAgeUnknown + reportRecord.Listeners
			c.StreamsGenderMaleAgeUnknown = c.StreamsGenderMaleAgeUnknown + reportRecord.Streams
		}
	} else if reportRecord.Gender == "2" {
		switch reportRecord.AgeBand {
		case "<18":
			c.ListenersGenderFemaleAge0_17 = c.ListenersGenderFemaleAge0_17 + reportRecord.Listeners
			c.StreamsGenderFemaleAge0_17 = c.StreamsGenderFemaleAge0_17 + reportRecord.Streams
		case "18-24":
			c.ListenersGenderFemaleAge18_24 = c.ListenersGenderFemaleAge18_24 + reportRecord.Listeners
			c.StreamsGenderFemaleAge18_24 = c.StreamsGenderFemaleAge18_24 + reportRecord.Streams
		case "25-34":
			c.ListenersGenderFemaleAge25_34 = c.ListenersGenderFemaleAge25_34 + reportRecord.Listeners
			c.StreamsGenderFemaleAge25_34 = c.StreamsGenderFemaleAge25_34 + reportRecord.Streams
		case "35-44":
			c.ListenersGenderFemaleAge35_44 = c.ListenersGenderFemaleAge35_44 + reportRecord.Listeners
			c.StreamsGenderFemaleAge35_44 = c.StreamsGenderFemaleAge35_44 + reportRecord.Streams
		case "45-54":
			c.ListenersGenderFemaleAge45_54 = c.ListenersGenderFemaleAge45_54 + reportRecord.Listeners
			c.StreamsGenderFemaleAge45_54 = c.StreamsGenderFemaleAge45_54 + reportRecord.Streams
		case "55-64":
			c.ListenersGenderFemaleAge55_64 = c.ListenersGenderFemaleAge55_64 + reportRecord.Listeners
			c.StreamsGenderFemaleAge55_64 = c.StreamsGenderFemaleAge55_64 + reportRecord.Streams
		case "65+":
			c.ListenersGenderFemaleAge65Plus = c.ListenersGenderFemaleAge65Plus + reportRecord.Listeners
			c.StreamsGenderFemaleAge65Plus = c.StreamsGenderFemaleAge65Plus + reportRecord.Streams
		default:
			c.ListenersGenderFemaleAgeUnknown = c.ListenersGenderFemaleAgeUnknown + reportRecord.Listeners
			c.StreamsGenderFemaleAgeUnknown = c.StreamsGenderFemaleAgeUnknown + reportRecord.Streams
		}
	} else {
		switch reportRecord.AgeBand {
		case "<18":
			c.ListenersGenderUnknownAge0_17 = c.ListenersGenderUnknownAge0_17 + reportRecord.Listeners
			c.StreamsGenderUnknownAge0_17 = c.StreamsGenderUnknownAge0_17 + reportRecord.Streams
		case "18-24":
			c.ListenersGenderUnknownAge18_24 = c.ListenersGenderUnknownAge18_24 + reportRecord.Listeners
			c.StreamsGenderUnknownAge18_24 = c.StreamsGenderUnknownAge18_24 + reportRecord.Streams
		case "25-34":
			c.ListenersGenderUnknownAge25_34 = c.ListenersGenderUnknownAge25_34 + reportRecord.Listeners
			c.StreamsGenderUnknownAge25_34 = c.StreamsGenderUnknownAge25_34 + reportRecord.Streams
		case "35-44":
			c.ListenersGenderUnknownAge35_44 = c.ListenersGenderUnknownAge35_44 + reportRecord.Listeners
			c.StreamsGenderUnknownAge35_44 = c.StreamsGenderUnknownAge35_44 + reportRecord.Streams
		case "45-54":
			c.ListenersGenderUnknownAge45_54 = c.ListenersGenderUnknownAge45_54 + reportRecord.Listeners
			c.StreamsGenderUnknownAge45_54 = c.StreamsGenderUnknownAge45_54 + reportRecord.Streams
		case "55-64":
			c.ListenersGenderUnknownAge55_64 = c.ListenersGenderUnknownAge55_64 + reportRecord.Listeners
			c.StreamsGenderUnknownAge55_64 = c.StreamsGenderUnknownAge55_64 + reportRecord.Streams
		case "65+":
			c.ListenersGenderUnknownAge65Plus = c.ListenersGenderUnknownAge65Plus + reportRecord.Listeners
			c.StreamsGenderUnknownAge65Plus = c.StreamsGenderUnknownAge65Plus + reportRecord.Streams
		default:
			c.ListenersGenderUnknownAgeUnknown = c.ListenersGenderUnknownAgeUnknown + reportRecord.Listeners
			c.StreamsGenderUnknownAgeUnknown = c.StreamsGenderUnknownAgeUnknown + reportRecord.Streams
		}
	}

	return c
}

func processContentDemographics(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket, date, licensor, vendorID string, content map[string]contentItem) {
	input := &s3.GetObjectInput{
		Bucket: aws.String(s3SourceBucket),
		Key:    aws.String(makeContentDemographicsSourceS3Key(date, licensor, vendorID)),
	}

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

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

	resultMap := make(map[string]contentDemographicsRecord)

	var headers []string
	firstItem := true

	scanner := bufio.NewScanner(reader)
	for scanner.Scan() {
		row := strings.Split(string(scanner.Text()), "\t")
		if firstItem == true {
			firstItem = false
			headers = row
		} else {
			reportRecord := parseReportRow(headers, row)
			contentItem := content[reportRecord.AppleIdentifier]
			countryCode := strings.ToLower(reportRecord.StorefrontName)

			if contentItem.ISRC != "" {
				trackKey := reportRecord.IngestDatestamp + "\xff" + reportRecord.StorefrontName + "\xff" + contentItem.ISRC

				t, ok := resultMap[trackKey]
				if ok == false {
					t = contentDemographicsRecord{
						ReportDate:     date,
						ReportVendorID: vendorID,
						Date:           reportRecord.IngestDatestamp,
						CountryCode:    countryCode,
						Licensor:       licensor,
						ISRC:           contentItem.ISRC,
					}
					resultMap[trackKey] = t
				}

				resultMap[trackKey] = mutateContentDemographics(t, reportRecord)
			}
		}
	}
	if err := scanner.Err(); err != nil {
		log.Fatal(err)
	}
	resultSession := session.Must(session.NewSession(&aws.Config{Region: aws.String(s3TargetService.SigningRegion), S3UseAccelerate: aws.Bool(false)}))
	resultUploader := s3manager.NewUploader(resultSession)
	resultReader, resultWriter := io.Pipe()

	go func() {
		gw := gzip.NewWriter(resultWriter)
		for _, v := range resultMap {
			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()
		resultWriter.Close()
	}()

	resultResponse, err := resultUploader.Upload(&s3manager.UploadInput{
		Bucket:          aws.String(s3TargetBucket),
		Key:             aws.String(makeContentDemographicsTargetS3Key(date, licensor, vendorID)),
		Body:            resultReader,
		ContentType:     aws.String("application/x-ndjson"),
		ContentEncoding: aws.String("gzip"),
		ACL:             aws.String("bucket-owner-full-control"),
	})
	if err != nil {
		log.Fatal(err)
	}
	_ = resultResponse
}

func runContentDemographicsTask(s3SourceService *s3.S3, s3TargetService *s3.S3, s3SourceBucket, s3TargetBucket, date, licensor, vendorID string) {
	targetExist := targetFileExists(s3TargetService, s3TargetBucket, makeContentDemographicsTargetS3Key(date, licensor, vendorID))
	if targetExist == true {
		log.Println(date, licensor, vendorID, "already processed")
		return
	}

	contentExist := sourceFileExists(s3SourceService, s3SourceBucket, makeContentSourceS3Key(date, licensor, vendorID))
	if contentExist == false {
		log.Println(date, licensor, vendorID, "content doesn't exist")
		return
	}

	contentDemographicsExist := sourceFileExists(s3SourceService, s3SourceBucket, makeContentDemographicsSourceS3Key(date, licensor, vendorID))
	if contentDemographicsExist == false {
		log.Println(date, licensor, vendorID, "contentdemographics doesn't exist")
		return
	}

	log.Println(date, licensor, vendorID, "fetching content")
	content := getContent(s3SourceService, s3SourceBucket, date, licensor, vendorID)

	log.Println(date, licensor, vendorID, "starting contentdemographics")
	processContentDemographics(s3SourceService, s3TargetService, s3SourceBucket, s3TargetBucket, date, licensor, vendorID, content)
}

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

	for d := range dates {
		date := dates[d]
		for l := range licensors {
			licensor := licensors[l]

			for _, vendorID := range VendorMap[licensor] {
				runContentDemographicsTask(s3SourceService, s3TargetService, s3Sourcebucket, s3TargetBucket, date, licensor, vendorID)
			}
		}
	}
}