package report

import (
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/pingcap/log"
	"github.com/tidbcloud/tiinf/sdk/measure"
	"go.uber.org/zap"
)

type BillingReport struct {
	CostID        string
	TenantID      uint64
	Sku           string
	Cost          int64
	CostTimeStamp int64
	Labels        map[string]string
	ShouldLog     bool
}

func (r *BillingReport) DeepCopy() *BillingReport {
	result := *r
	result.Labels = make(map[string]string, len(r.Labels))
	for k, v := range r.Labels {
		result.Labels[k] = v
	}
	return &result
}

// SQSReporter wraps the MeasureSQSImpl from the measure package to provide
// additional functionality specific to SQS reporting.
type SQSReporter struct {
	quit        chan struct{}
	sqs         *measure.MeasureSQSImpl
	mu          sync.Mutex
	cacheReport map[uint64]map[string]measure.BillingMeasure
}

// NewSQSReporter creates a new instance of SQSReporter based on the provided
// environment string. It returns a pointer to the SQSReporter and an error
// if the initialization fails.
//
// Parameters:
// - env: A string representing the environment "prod"/"staging"/"dev"
func NewSQSReporter(env string) (*SQSReporter, error) {
	sqs := &measure.MeasureSQSImpl{}
	if err := sqs.InitMeasureSqsHandler(env, ""); err != nil {
		return nil, err
	}
	r := &SQSReporter{
		quit:        make(chan struct{}),
		sqs:         sqs,
		cacheReport: make(map[uint64]map[string]measure.BillingMeasure),
	}
	go r.sendMeasureLoop()
	return r, nil
}

func (r *SQSReporter) Report(report BillingReport) error {
	bm := measure.BillingMeasure{
		TenantID:      report.TenantID,
		CostID:        report.CostID,
		Cost:          report.Cost,
		CostTimestamp: report.CostTimeStamp,
		Labels:        report.Labels,
		Unit:          report.Sku,
	}
	if err := r.sqs.CheckMeasure(bm); err != nil {
		return err
	}
	key := measureKey(bm)
	r.mu.Lock()
	defer r.mu.Unlock()
	if _, ok := r.cacheReport[bm.TenantID]; !ok {
		r.cacheReport[bm.TenantID] = make(map[string]measure.BillingMeasure)
		r.cacheReport[bm.TenantID][key] = bm
	} else {
		if old, ok := r.cacheReport[bm.TenantID][key]; ok {
			bm.Cost += old.Cost
		}
		r.cacheReport[bm.TenantID][key] = bm
	}
	return nil
}

func measureKey(m measure.BillingMeasure) string {
	return fmt.Sprintf("%s;%d;%s", m.Unit, m.TenantID, mapKey(m.Labels))
}

func mapKey(m map[string]string) string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)

	var sb strings.Builder
	for i, k := range keys {
		sb.WriteString(k)
		sb.WriteString(":")
		sb.WriteString(m[k])
		if i < len(keys)-1 {
			sb.WriteString(";")
		}
	}
	return sb.String()
}

const retryTimes = 20
const sendMeasureInterval = time.Minute * 10

func (r *SQSReporter) sendMeasureLoop() {
	ticker := time.NewTicker(sendMeasureInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			r.sendMeasures()
		case <-r.quit:
			return
		}
	}
}

// Close stops the SQSReporter from sending measures.
func (r *SQSReporter) Close() {
	close(r.quit)
}

func (r *SQSReporter) sendMeasures() {
	r.mu.Lock()
	reports := make(map[uint64][]measure.BillingMeasure)
	for _, measure := range reports {
		for _, m := range measure {
			reports[m.TenantID] = append(reports[m.TenantID], m)
		}
	}
	r.cacheReport = make(map[uint64]map[string]measure.BillingMeasure)
	r.mu.Unlock()

	var succCount int
	for i := 1; i <= retryTimes; i++ {
		failed := make(map[uint64][]measure.BillingMeasure)
		for tenantID, measures := range reports {
			for _, m := range measures {
				if err := r.sqs.SendMeasure(m); err != nil {
					log.Error("failed to send measure", zap.Uint64("tenantID", tenantID), zap.Error(err), zap.Any("measure", m))
					failed[tenantID] = append(failed[tenantID], m)
				} else {
					succCount++
				}
			}
		}
		if len(failed) == 0 {
			break
		}
		time.Sleep(time.Second * time.Duration(i))
		reports = failed
	}

	log.Info("send measures", zap.Int("success", succCount))

	if len(reports) > 0 {
		log.Info("some measures are failed to send", zap.Any("reports", reports))
	}
}