package cluster

import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
	"strconv"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/shared-provider/pkg/types"
	apitypes "k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/clock"
	"k8s.io/client-go/util/retry"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

const defaultAccumulatePeriodDuration = time.Minute * 10

// keys of label we use to save bandwidth info.
const (
	trafficUsageAnnKey = "bandwidth.tidbcloud.com/usage"
	// UpdateBandwidthDurationAnnKey set the duration to update info.
	// only used for test.
	UpdateBandwidthDurationAnnKey = "bandwidth.tidbcloud.com/update-duration"
)

// once reach TotalInBytes we will limit the bandwidth in LimitInBytes.
const (
	TotalInBytes uint64 = 20 * 1024 * 1024 * 1024
	LimitInBytes uint64 = 10 * 1024
)

type TrafficUsageInfo struct {
	IngressUsage       float64   `json:"ingressUsage"`
	EgressUsage        float64   `json:"egressUsage"`
	LastAccumulateTime time.Time `json:"lastAccumulateTime"`
}

type BandwidthHandler struct {
	client      client.Client
	infraClient client.Client
	clock       clock.Clock
}

func NewBandwidthHandler(client client.Client, infraClient client.Client) *BandwidthHandler {
	return &BandwidthHandler{
		client:      client,
		infraClient: infraClient,
		clock:       &clock.RealClock{},
	}
}

// timePeriod return the start and end time period this time belong to.
// we split the time per week per period, and a period start at Monday 00:00.
func timePeriod(t time.Time) (start time.Time, end time.Time) {
	year, month, day := t.Date()

	start = time.Date(year, month, day, 0, 0, 0, 0, t.Location())

	daysPerWeek := 7
	// t.Weekday() return a Weekday specifies a day of the week (Sunday = 0, ...).
	// adjust to let it be (Monday = 1, ...).
	weekDay := (int(t.Weekday()) - 1 + daysPerWeek) % daysPerWeek
	start = start.AddDate(0, 0, -weekDay)

	end = start.AddDate(0, 0, daysPerWeek)

	return
}

func (h *BandwidthHandler) syncAnnToTC(ctx context.Context, cc *cloudv1.Cluster, tc *v1alpha1.TidbCluster) error {
	if cc.Annotations[types.AnnoKeyIngress] == tc.Annotations[types.AnnoKeyIngress] &&
		cc.Annotations[types.AnnoKeyEgress] == tc.Annotations[types.AnnoKeyEgress] {
		return nil
	}

	mapRecord := NewMapRecord(tc.Annotations)
	v, ok := cc.Annotations[types.AnnoKeyIngress]
	if ok {
		mapRecord.Set(types.AnnoKeyIngress, v)
		mapRecord.Set(types.AnnoKeyIngressPort, "4000")
	} else {
		mapRecord.Delete(types.AnnoKeyIngress)
		mapRecord.Delete(types.AnnoKeyIngressPort)
	}

	v, ok = cc.Annotations[types.AnnoKeyEgress]
	if ok {
		mapRecord.Set(types.AnnoKeyEgress, v)
		mapRecord.Set(types.AnnoKeyEgressPort, "4000")
	} else {
		mapRecord.Delete(types.AnnoKeyEgress)
		mapRecord.Delete(types.AnnoKeyEgressPort)
	}

	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
		err := h.client.Get(ctx, apitypes.NamespacedName{Namespace: tc.GetNamespace(), Name: tc.GetName()}, tc)
		if err != nil {
			return err
		}

		tc.Annotations = mapRecord.Apply(tc.Annotations)

		return h.client.Update(ctx, tc)
	})

	return err
}

func getAccumulatePeriodDuration(cc *cloudv1.Cluster) time.Duration {
	if str, ok := cc.Annotations[UpdateBandwidthDurationAnnKey]; ok {
		t, err := time.ParseDuration(str)
		if err != nil {
			log.Log.Error(err, "wrong format", "value", str)
			return defaultAccumulatePeriodDuration
		}
		return t
	}

	return defaultAccumulatePeriodDuration
}

// Sync query metrics from Querier and update label and condition for bandwidth throttle.
// It is expected to resyn after the return checkAfter duration.
// nolint: funlen // Function 'Sync' has too many statements
func (h *BandwidthHandler) Sync(
	ctx context.Context,
	cc *cloudv1.Cluster,
	tc *v1alpha1.TidbCluster,
	querier Querier,
) (checkAfter time.Duration, err error) {
	logger := log.FromContext(ctx)

	err = h.syncAnnToTC(ctx, cc, tc)
	if err != nil {
		return 0, err
	}

	now := h.clock.Now().In(time.UTC)
	start, _ := timePeriod(now)
	mapRecord := NewMapRecord(cc.Annotations)

	trafficUsage := new(TrafficUsageInfo)
	data, ok := mapRecord.Get(trafficUsageAnnKey)
	if ok {
		err := json.Unmarshal([]byte(data), trafficUsage)
		if err != nil {
			return 0, errors.AddStack(err)
		}
	}

	logger.V(2).Info("start sync bandwidth", "now", now, "lastAccumulateTime", trafficUsage.LastAccumulateTime)

	accumulatePeriodDuration := getAccumulatePeriodDuration(cc)
	if now.Sub(trafficUsage.LastAccumulateTime) < accumulatePeriodDuration {
		checkAfter := accumulatePeriodDuration - now.Sub(trafficUsage.LastAccumulateTime)
		logger.V(2).Info("early return to re-sync", "after", checkAfter)
		return checkAfter, nil
	}

	if trafficUsage.LastAccumulateTime.Before(start) {
		trafficUsage.LastAccumulateTime = start
		trafficUsage.IngressUsage = 0
		trafficUsage.EgressUsage = 0
	}

	// get bytes used in (LastAccumulateTime, now]
	v := make(map[string]string)
	base10 := 10
	duration := strconv.FormatInt(int64(now.Sub(trafficUsage.LastAccumulateTime).Seconds()), base10) + "s"
	v["query"] = fmt.Sprintf("sum by (type) (increase(tidb_server_packet_io_bytes_sum[%s]))", duration)
	v["time"] = now.Format(time.RFC3339)
	// a dummy value just avoid cache
	v["_"] = strconv.FormatInt(time.Now().Unix(), base10)
	body, err := querier.Query(ctx, "/api/v1/query", v)
	if err != nil {
		return 0, errors.AddStack(err)
	}

	logger.V(2).Info("query prometheus", "params", v, "response", string(body))

	response := new(QueryResponse)
	err = json.Unmarshal(body, response)
	if err != nil {
		return 0, errors.AddStack(err)
	}

	toFloat := func(v interface{}) (value float64, err error) {
		rv := reflect.ValueOf(v)
		value, err = strconv.ParseFloat(rv.String(), 64)
		if err != nil {
			return 0, errors.Annotatef(err, "failed to parse %q", rv.String())
		}
		return
	}

	for _, result := range response.Data.Result {
		validValueLen := 2 // timestamp and value
		if result.Metric["type"] == "write" {
			if len(result.Value) < validValueLen {
				return 0, errors.Errorf("invalid response %q", body)
			}

			v, err := toFloat(result.Value[1])
			if err != nil {
				return 0, err
			}

			trafficUsage.EgressUsage += v
		} else if result.Metric["type"] == "read" {
			if len(result.Value) < validValueLen {
				return 0, errors.Errorf("invalid response %q", body)
			}

			v, err := toFloat(result.Value[1])
			if err != nil {
				return 0, err
			}

			trafficUsage.IngressUsage += v
		}
	}
	trafficUsage.LastAccumulateTime = now

	bytes, err := json.Marshal(trafficUsage)
	if err != nil {
		return 0, errors.AddStack(err)
	}
	mapRecord.Set(trafficUsageAnnKey, string(bytes))
	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		err := h.infraClient.Get(ctx, apitypes.NamespacedName{Namespace: cc.GetNamespace(), Name: cc.GetName()}, cc)
		if err != nil {
			return err
		}

		cc.Annotations = mapRecord.Apply(cc.Annotations)

		return h.infraClient.Update(ctx, cc)
	})
	if err != nil {
		return 0, errors.AddStack(err)
	}

	return accumulatePeriodDuration, nil
}

type mapRecordOP struct {
	key   string
	value *string
}

// MapRecord will record operation on a map and ca re-apply at another map.
type MapRecord struct {
	mp  map[string]string
	ops []*mapRecordOP
}

func NewMapRecord(mp map[string]string) *MapRecord {
	return &MapRecord{
		mp: mp,
	}
}

func (m *MapRecord) Set(k string, v string) {
	m.ops = append(m.ops, &mapRecordOP{key: k, value: &v})
	if m.mp == nil {
		m.mp = make(map[string]string)
	}
	m.mp[k] = v
}

func (m *MapRecord) Delete(k string) {
	m.ops = append(m.ops, &mapRecordOP{key: k, value: nil})
	delete(m.mp, k)
}

func (m *MapRecord) Get(k string) (v string, ok bool) {
	v, ok = m.mp[k]
	return
}

func (m *MapRecord) Apply(mp map[string]string) map[string]string {
	for _, op := range m.ops {
		if op.value == nil {
			delete(mp, op.key)
		} else {
			if mp == nil {
				mp = make(map[string]string)
			}
			mp[op.key] = *op.value
		}
	}

	return mp
}