package cost

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

	"github.com/google/uuid"
	"github.com/samber/lo"
	"github.com/tidbcloud/core-svc/event"
	"github.com/tidbcloud/cost-collect/pkg/cloud/gcp"
	"github.com/tidbcloud/cost-collect/pkg/global"
	"github.com/tidbcloud/pkgv2/asserts"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/zap"
)

const (
	// GCPSQLV2Template is used for pre-aggregated table to speedup query speed and reduce cost
	GCPSQLV2Template = `
SELECT
    region,amount,unit,description,service_name,cost,
    user_label_tenant_id,user_label_project_id,user_label_cluster_id,user_label_environment_id,user_label_component_id,
    user_label_usedby,user_label_servicetype
FROM ` + "`" + `{{ .PreAggregatedGCPTable }}` + "`" + `
WHERE
	usage_start_date >= DATE("{{ .StartTime }}") AND usage_start_date < DATE("{{ .EndTime }}")
    {{ if .TenantID }}AND user_label_tenant_id = "{{ .TenantID }}"{{ end }}
    {{ if .ProjectID }}AND user_label_project_id = "{{ .ProjectID }}"{{ end }}
ORDER BY user_label_tenant_id,user_label_project_id
;`
	// AWSSQLV2Template is used for pre-aggregated table to speedup query speed and reduce cost
	AWSSQLV2Template = `
SELECT resource_tags_user_tenant,resource_tags_user_project,resource_tags_user_usedby,
    line_item_blended_cost,line_item_unblended_cost,line_item_usage_amount,pricing_unit,
    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 ` + "`" + `{{ .PreAggregatedAWSTable }}` + "`" + `
WHERE 
	line_item_usage_start_date >= DATE("{{ .StartTime }}") AND line_item_usage_start_date < DATE("{{ .EndTime }}")
    {{ if .TenantID }}AND resource_tags_user_tenant = '{{ .TenantID }}'{{ end }}
    {{ if .ProjectID }}AND resource_tags_user_project = '{{ .ProjectID }}'{{ end }}
ORDER BY resource_tags_user_tenant,resource_tags_user_project
`
)

var billtemplatefns = template.FuncMap{
	"sub": func(y, x int) int {
		return x - y
	},
	"StringsJoin": strings.Join,
}

func GetProviderCost(ctx context.Context, request global.Request, gcpClient *gcp.BigQueryClient) ([]global.Item, error) {
	var items []global.Item
	gcpItems, err := getGCPNetworkBillData(ctx, request, gcpClient)
	if err != nil {
		return nil, err
	}
	items = append(items, gcpItems...)
	awsItems, err := getAWSNetworkBillData(ctx, request, gcpClient)
	if err != nil {
		return nil, err
	}
	items = append(items, awsItems...)
	return items, nil
}

func (t Manager) streamProviderCost(ctx context.Context, request global.Request) error {
	items, err := GetProviderCost(ctx, request, t.gcpClient)
	if err != nil {
		return err
	}

	return t.groupAndPublishItems(ctx, request, items)
}

func (t Manager) groupAndPublishItems(ctx context.Context, request global.Request, items []global.Item) error {
	const (
		// finalizerProjectID represents all project metering data are published into eventbus in this `batch`. The `batch`
		// is a group project metering data which have the same `batch_id`.
		finalizerProjectID = "0"
	)
	logger := log.WithContextL(ctx)

	projectID2Items := make(map[string][]global.Item, len(items))
	for _, item := range items {
		projectID2Items[item.Labels.ProjectID] = append(projectID2Items[item.Labels.ProjectID], item)
	}

	batchID := uuid.New().String()
	logger = logger.With(zap.String("batchID", batchID))
	logger.Info("try to publish data", zap.Int("projects", len(projectID2Items)))

	var hasError bool
	for projectID, itemsOfProject := range projectID2Items {
		if err := t.eventPublisher(ctx, request, batchID, projectID, itemsOfProject); err != nil {
			logger.Error("publish2eventbus failed",
				zap.String("projectID", projectID),
				zap.Int("itemsOfProject", len(itemsOfProject)),
				zap.Error(err))
			hasError = true
			continue
		}
	}
	if !hasError {
		// If all project metering records are successfully published,
		// we publish a special project metering record to indicate this batch end.
		return t.eventPublisher(ctx, request, batchID, finalizerProjectID, nil)
	} else {
		logger.Error("some error happen in this batch, later retry is needed")
		return fmt.Errorf("some errors happened while publishing project metering data")
	}
}

func (t Manager) publish2eventbusImp(ctx context.Context, request global.Request, batchID, projectID string, items []global.Item) error {
	logger := log.WithContextL(ctx).With(zap.String("batchID", batchID))
	pid, err := strconv.ParseUint(projectID, 10, 64)
	if err != nil {
		logger.Error("parse projectID failed", zap.Error(err), zap.String("projectID", projectID))
		return nil
	}
	startTime, err := time.ParseInLocation(global.TimeTemplate, request.StartTime, time.UTC)
	asserts.BeSuccess(err)
	endTime, err := time.ParseInLocation(global.TimeTemplate, request.EndTime, time.UTC)
	asserts.BeSuccess(err)

	meteringData := &event.ProjectMeteringData{
		ProjectID: pid,
		StartTime: startTime.Unix(),
		EndTime:   endTime.Unix(),
		BatchID:   batchID,
		Items:     nil, // fill later
	}
	for _, item := range items {
		meteringData.Items = append(meteringData.Items, event.ProjectMeteringDataItem{
			Amount:          item.Amount,
			Cost:            item.Cost,
			Labels:          item.Labels.ToMap(),
			ServiceName:     item.ServiceName,
			ResourceType:    item.ResourceType,
			ResourceSubType: item.ResourceSubtype,
			Unit:            item.Unit,
			DataVersion:     item.DataVersion,
		})
	}
	ev := &tieventbus.Event{
		ID:   t.buildEventID(),
		Type: string(event.TopicProjectMeteringData),
		Data: meteringData,
	}

	return t.eventbus.Publish(ev)
}

func (t Manager) buildEventID() string {
	return uuid.New().String()
}

func getGCPNetworkBillData(ctx context.Context, request global.Request, gcpClient *gcp.BigQueryClient) ([]global.Item, error) {
	logger := log.WithContextL(ctx)
	logger.Sugar().Infof("getGCPNetworkBillData, request=%+v", request)

	sql, err := CreateBigquerySQL(ctx, gcpClient.Config.GCPBillTemplate,
		gcp.BigQueryParam{
			Request: request,
			Config:  gcpClient.Config,
		})
	if err != nil {
		return nil, err
	}
	rawRows, err := gcpClient.ExecQuery(ctx, sql)
	if err != nil {
		return nil, err
	}
	var items []global.Item
	for _, row := range rawRows {
		item, err := gcpDataExtract(row, request)
		if err != nil {
			return nil, err
		}
		items = append(items, *item)
	}
	return items, nil
}

func getAWSNetworkBillData(ctx context.Context, request global.Request, gcpClient *gcp.BigQueryClient) ([]global.Item, error) {
	logger := log.WithContextL(ctx)
	logger.Sugar().Infof("getAWSNetworkBillData, request=%+v", request)

	sql, err := CreateBigquerySQL(ctx, gcpClient.Config.AWSBillTemplate,
		gcp.BigQueryParam{
			Request: request,
			Config:  gcpClient.Config,
		})
	if err != nil {
		return nil, err
	}
	rawRows, err := gcpClient.ExecQuery(ctx, sql)
	if err != nil {
		return nil, err
	}

	// Only for debug
	logger.Info("aws network bill data details",
		zap.Uint64("TenantID", request.TenantID),
		zap.Uint64("ProjectID", request.ProjectID),
		zap.String("StartTime", request.StartTime),
		zap.String("EndTime", request.EndTime),
		zap.Any("details", rawRows),
	)

	var out []global.Item
	for _, row := range rawRows {
		items, err := awsDataExtract(row, request)
		if err != nil {
			return nil, err
		}

		out = append(out, lo.Map(items, func(item *global.Item, index int) global.Item { return *item })...)
	}
	return out, nil
}

func GetAWSSQLTemplate() (*template.Template, error) {
	return template.New("bigquery").Funcs(billtemplatefns).Parse(AWSSQLV2Template)
}

func GetGCPSQLTemplate() (*template.Template, error) {
	return template.New("bigquery").Funcs(billtemplatefns).Parse(GCPSQLV2Template)
}

func CreateBigquerySQL(ctx context.Context, tpl *template.Template, args gcp.BigQueryParam) (string, error) {
	logger := log.WithContextL(ctx)
	buf := new(bytes.Buffer)
	err := tpl.Execute(buf, args)
	sql := buf.String()
	logger.Info("generated SQL", zap.String("sql", sql))
	return sql, err
}