package aws

import (
	"bytes"
	"context"
	"fmt"
	"strconv"
	"strings"
	"text/template"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/athena"
	"github.com/pingcap/log"
	"github.com/tidbcloud/cost-collect/pkg/global"
)

const SQLTemplateV2 = `SELECT "resource_tags_user_tenant","resource_tags_user_project","resource_tags_user_usedby",
	sum("line_item_blended_cost") as "line_item_blended_cost",sum("line_item_unblended_cost") as "line_item_unblended_cost",sum("line_item_usage_amount") as "line_item_usage_amount","pricing_unit",
    "line_item_resource_id",CONCAT("line_item_product_code",':',"product_servicename") AS "service_name","line_item_operation","product_region_code",
	"resource_tags_user_cluster","resource_tags_user_component","resource_tags_user_environment","resource_tags_user_servicetype"
FROM "{{ .Database }}"."{{ .Table }}"
WHERE "line_item_usage_start_date" >= timestamp '{{ .StartTime }}' AND "line_item_usage_start_date" < timestamp '{{ .EndTime }}'
    AND (
		"line_item_product_code" IN ('AWSELB','AmazonS3') OR 
	    ("line_item_product_code"='AmazonEC2' AND "product_servicename" = 'AWS Data Transfer')
	)
	AND "line_item_unblended_cost" > 0
	AND "resource_tags_user_usedby" LIKE 'customer%'
	AND "resource_tags_user_component" != 'extra-tidb'
    AND "resource_tags_user_tenant" = '{{ .TenantID }}'
	{{ if .ProjectID }}AND "resource_tags_user_project" = '{{ .ProjectID }}'{{ end }}
GROUP BY "resource_tags_user_tenant","resource_tags_user_project","resource_tags_user_usedby",
"pricing_unit",
"line_item_resource_id","line_item_product_code","product_servicename","line_item_operation","product_region_code",
"resource_tags_user_cluster","resource_tags_user_component","resource_tags_user_environment","resource_tags_user_servicetype"
;
`

const SQLTemplate = `SELECT "resource_tags_user_tenant","resource_tags_user_project","resource_tags_user_usedby",
	sum("line_item_blended_cost") as "line_item_blended_cost", sum("line_item_unblended_cost") as "line_item_unblended_cost", sum("line_item_usage_amount") as "line_item_usage_amount","pricing_unit",
    "line_item_resource_id",CONCAT("line_item_product_code",':',"product_servicename") AS "service_name","line_item_operation","product_region_code",
	"resource_tags_user_cluster","resource_tags_user_component","resource_tags_user_environment","resource_tags_user_servicetype"
FROM "{{ .Database }}"."{{ .Table }}"
WHERE "line_item_usage_start_date" >= timestamp '{{ .StartTime }}' AND "line_item_usage_start_date" < timestamp '{{ .EndTime }}'
    AND "line_item_product_code" IN ('AWSELB','AmazonS3')
	AND "resource_tags_user_usedby" LIKE 'customer%'
    AND "resource_tags_user_tenant" = '{{ .TenantID }}'
	{{ if .ProjectID }}AND "resource_tags_user_project" = '{{ .ProjectID }}'{{ end }}
GROUP BY "resource_tags_user_tenant","resource_tags_user_project","resource_tags_user_usedby", "pricing_unit",
"line_item_resource_id","line_item_product_code","product_servicename","line_item_operation","product_region_code",
"resource_tags_user_cluster","resource_tags_user_component","resource_tags_user_environment","resource_tags_user_servicetype"
;
`

const (
	DataVersionV1 = "v1"
	DataVersionV2 = "v2"
)

type AthenaParam struct {
	global.Request
	AthenaSharedParam
}

type AthenaSharedParam struct {
	Database string
	Table    string
}
type AWSBillingClient struct {
	client       *athena.Athena
	outputBucket string // output to s3
	waitTime     int    //sleep for wait reqeust

	sharedParam AthenaSharedParam
}

func NewBillingClient(outputBucket string, waitTime int, sharedParam AthenaSharedParam) (*AWSBillingClient, error) {
	if outputBucket == "" ||
		sharedParam.Database == "" || sharedParam.Table == "" {
		err := fmt.Errorf("missing aws athena parameter")
		return nil, err
	}

	/*
		cred := credentials.NewStaticCredentials(
			os.Getenv("AWS_ACCESS_KEY_ID"),     // Access key
			os.Getenv("AWS_SECRET_ACCESS_KEY"), // secret
			"",
		)
	*/
	conf := aws.Config{
		//Region: aws.String(os.Getenv("AWS_DEFAULT_REGION")),
		//Credentials: cred,
	}
	sess, err := session.NewSession(&conf)
	if err != nil {
		return nil, err
	}
	athenaClient := athena.New(sess)
	client := AWSBillingClient{
		client:       athenaClient,
		outputBucket: outputBucket,
		waitTime:     waitTime,
		sharedParam:  sharedParam,
	}

	return &client, nil
}

func (t AWSBillingClient) query(sql string) ([]*athena.Row, error) {

	resultConf := &athena.ResultConfiguration{
		OutputLocation: aws.String(t.outputBucket),
	}
	params := &athena.StartQueryExecutionInput{
		QueryString:         aws.String(sql),
		ResultConfiguration: resultConf,
	}

	// create exec job
	req, output := t.client.StartQueryExecutionRequest(params)

	err := req.Send()
	if err != nil {
		log.S().Errorf("RequestError: err:%s", err)
		return nil, err
	}

	log.S().Debugf("QuerySQL:%s", sql)
	log.S().Debugf("Query Execution Response ID: %s", *output.QueryExecutionId)

	var MaxResults int64 = 1000

	params2 := &athena.GetQueryResultsInput{
		MaxResults:       &MaxResults,
		QueryExecutionId: output.QueryExecutionId,
	}

	// wait job ready
	rows := []*athena.Row{}
	nextToekn := ""
	for {
		// Too much data requires pagination
		if nextToekn != "" {
			params2 = params2.SetNextToken(nextToekn)
			nextToekn = ""
		} else {
			time.Sleep(time.Second * time.Duration(t.waitTime))
		}

		req2, output2 := t.client.GetQueryResultsRequest(params2)
		err = req2.Send()

		if err != nil {
			if strings.HasPrefix(err.Error(), "InvalidRequestException:") &&
				(strings.Contains(err.Error(), "Current state: QUEUED") || strings.Contains(err.Error(), "Current state: RUNNING")) {

				log.S().Debugf("taskwait:%s,%s", *output.QueryExecutionId, err)
				continue
			}

			log.S().Errorf("Request err:%s", err)

			return nil, err
		}
		rows = append(rows, output2.ResultSet.Rows...)

		if output2.NextToken != nil {
			nextToekn = *output2.NextToken
			log.S().Infof("rows has more page,rows now %d", len(rows))
			continue
		}
		break
	}
	return rows, nil
}

func (t AWSBillingClient) GetProviderCost(ctx context.Context, request global.Request) ([]global.Item, error) {
	rows, err := t.ExecProviderCostQuery(ctx, request)
	if err != nil {
		return nil, err
	}
	schema := []string{}
	for _, v := range rows[0].Data {
		schema = append(schema, *v.VarCharValue)
	}
	items := []global.Item{}
	for _, row := range rows[1:] {

		// load label
		item, err := dataExtract(schema, row.Data, request)
		if err != nil {
			return nil, err
		}
		items = append(items, *item)
	}
	return items, nil
}

func (t AWSBillingClient) ExecProviderCostQuery(ctx context.Context, request global.Request) ([]*athena.Row, error) {
	args := AthenaParam{
		Request:           request,
		AthenaSharedParam: t.sharedParam,
	}
	sql, err := CreateAthenaSQL(args)
	if err != nil {
		return nil, err
	}

	rows, err := t.query(sql)
	if err != nil {
		return nil, err
	}
	return rows, nil
}

// dataExtract let map
func dataExtract(fileds []string, data []*athena.Datum, request global.Request) (*global.Item, error) {
	item := global.Item{}

	labels := map[string]string{}

	labelPrefix := "resource_tags_user_"

	// startTimeKey := "line_item_usage_start_date"
	// endTimeKey := "line_item_usage_end_date"
	amountKey := "line_item_usage_amount"

	//blendedCost := "line_item_blended_cost"
	unblendedCost := "line_item_unblended_cost"
	costKey := unblendedCost

	var err error

	item.StartTime = request.StartTime
	item.EndTime = request.EndTime

	for idx, name := range fileds {
		value := *data[idx].VarCharValue
		item.DataVersion = DataVersionV1
		if request.IsDataVersionV2() {
			item.DataVersion = DataVersionV2
		}

		if strings.HasPrefix(name, labelPrefix) {
			key := strings.TrimPrefix(name, labelPrefix)
			if key != "usedby" && key != "servicetype" {
				key += "_id"
			}
			labels[key] = value
		} else {
			switch name {
			case costKey: // cost
				item.Cost, err = strconv.ParseFloat(value, 64)
				if err != nil {
					return nil, err
				}
			case amountKey: //amount
				item.Amount, err = strconv.ParseFloat(value, 64)
				if err != nil {
					return nil, err
				}
			case "pricing_unit":
				item.Unit = value
			case "service_name":
				item.ServiceName = value
			case "line_item_operation":
				item.Description = value
			case "product_region_code":
				item.Location = value
			}

		}
	}

	err = global.Coverage(labels, &item.Labels)
	if err != nil {
		return nil, err
	}

	if request.IsDataVersionV2() {
		// classify for biz flag
		if strings.HasPrefix(item.ServiceName, "AWSELB") || strings.HasPrefix(item.ServiceName, "AmazonEC2") {
			item.ResourceType = global.CloudTypeNetwork

			// for resource subtype
			if strings.HasPrefix(item.Description, "InterZone-") || strings.HasPrefix(item.Description, "VPCPeering-") {
				// cross az or use peering connect in same region
				item.ResourceSubtype = global.CloudNetworkSubTypeCrossZone
				//} else if strings.HasPrefix(item.Description, "    ") {
				// cross region
				//	item.ResourceSubtype = global.CloudNetworkSubTypeCrossRegion
			} else if strings.HasPrefix(item.Description, "LoadBalancing-PublicIP-") {
				// internet
				item.ResourceSubtype = global.CloudNetworkSubTypeInternet
			} else if strings.HasPrefix(item.Description, "LoadBalancing:") {
				// loadbalance process
				item.ResourceSubtype = global.CloudNetworkSubTypeLoadbalance
			}
		} else if strings.HasPrefix(item.ServiceName, "AmazonS3") {
			item.ResourceType = global.CloudTypeStorage
		}
	} else {
		// service type
		if strings.HasPrefix(item.ServiceName, "AWSELB") {
			item.ResourceType = global.CloudTypeNetwork
		} else if strings.HasPrefix(item.ServiceName, "AmazonS3") {
			item.ResourceType = global.CloudTypeStorage
		}
	}

	item.Labels.Provider = "aws"
	item.Labels.UsedBy = "customer" //compatibility old tags,it use customer-tidb
	return &item, nil
}

func CreateAthenaSQL(args AthenaParam) (string, error) {
	sql := SQLTemplate
	if args.Request.IsDataVersionV2() {
		sql = SQLTemplateV2
	}
	tpl, err := template.New("athena").Parse(sql)
	if err != nil {
		return "", err
	}

	buf := new(bytes.Buffer)
	err = tpl.Execute(buf, args)

	return buf.String(), err
}