package datasource

import (
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"unsafe"

	"github.com/VictoriaMetrics/fastcache"
	"github.com/gin-gonic/gin"
	"github.com/google/uuid"
	"github.com/pingcap/log"
	"github.com/prometheus/prometheus/prompb"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/api/common"
	error2 "github.com/pingcap-inc/clinic/pkg/api/error"
	"github.com/pingcap-inc/clinic/pkg/auth/session"
	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/db/model"
	"github.com/pingcap-inc/clinic/pkg/importer/metrics"
)

const (

	// system is clinic
	// sub-system is diagnosis
	metricsNameMaxQPS         = "clinic_diagnosis_active_qps_max"
	metricsNameActiveDuration = "clinic_diagnosis_active_qps_duration"
	metricsNameReport         = "clinic_diagnosis_active_qps_report"
)

// use seprate struct for easy unit testing.
type ActiveQPSSink struct {
	metricsSink *metrics.MetricSink
	// for easy unit testing.
	getDBaasClusterFn func(ctx *session.CtxUser, id string, orgID uint) (*db.Cluster, error)
	// use org_id to find org
	getDBaasOrgFn func(ctx *session.CtxUser, orgID uint) (*db.Organization, error)
	// tmp cache for tenant id with tenant name and cluster name
	clusterCache *fastcache.Cache
	tenantCache  *fastcache.Cache
}

type ActiveOPSSinkOption func(s *ActiveQPSSink) error

func WithGetClusterInfoFn(fn func(ctx *session.CtxUser, id string, orgID uint) (*db.Cluster, error)) ActiveOPSSinkOption {
	return func(s *ActiveQPSSink) error {
		s.getDBaasClusterFn = fn
		return nil
	}
}

func WithGetOrgInfoFn(fn func(ctx *session.CtxUser, externalID uint) (*db.Organization, error)) ActiveOPSSinkOption {
	return func(s *ActiveQPSSink) error {
		s.getDBaasOrgFn = fn
		return nil
	}
}

func NewActiveQPSSink(metricsSink *metrics.MetricSink, opts ...ActiveOPSSinkOption) (*ActiveQPSSink, error) {
	sink := &ActiveQPSSink{
		metricsSink:  metricsSink,
		clusterCache: fastcache.New(32 << 20),
		tenantCache:  fastcache.New(32 << 20),
	}
	for _, opt := range opts {
		if err := opt(sink); err != nil {
			return nil, err
		}
	}
	return sink, nil
}

/*
use BatchInsert instead
func (sink *ActiveQPSSink) InsertActiveQPS(uctx *session.CtxUser, param *InsertActiveQPSParam) error {
	dts, mts, err := sink.buildTimeSeries(uctx, param)
	if err != nil {
		return err
	}
	if err = sink.metricsSink.Sink(*dts); err != nil {
		log.Error("sink time series failed", zap.Error(err))
		return err
	}
	if err = sink.metricsSink.Sink(*mts); err != nil {
		log.Error("sink time series failed", zap.Error(err))
		return err
	}
	return nil
}
*/

func (sink *ActiveQPSSink) BatchInsertActiveQPS(uctx *session.CtxUser, params []InsertActiveQPSParam) error {
	durationTss := make([]prompb.TimeSeries, 0, len(params))
	maxQPSTss := make([]prompb.TimeSeries, 0, len(params))
	for _, param := range params {
		meta, err := sink.getClusterMeta(uctx, &param)
		if err != nil {
			return err
		}
		durationTss = append(durationTss, *param.ToActiveDurationSeries(meta))
		maxQPSTss = append(maxQPSTss, *param.ToMaxQPSSeries(meta))
	}
	if err := sink.metricsSink.SinkBatch(durationTss); err != nil {
		log.Error("batch insert failed", zap.Error(err))
		return err
	}
	if err := sink.metricsSink.SinkBatch(maxQPSTss); err != nil {
		log.Error("batch insert failed", zap.Error(err))
		return err
	}
	return nil
}

func (sink *ActiveQPSSink) BatchInsertActiveQPSRelatedReport(uctx *session.CtxUser, reportMap map[uint]*InsertActiveQPSParam) error {
	reportTss := make([]prompb.TimeSeries, 0, len(reportMap))
	for reportID, param := range reportMap {
		ts := &prompb.TimeSeries{
			Labels: []prompb.Label{
				{
					Name:  "__name__",
					Value: metricsNameReport,
				},
				{
					Name:  "cluster_id",
					Value: param.ClusterID,
				},
			},
			Samples: []prompb.Sample{
				{
					Timestamp: param.StartUnix * 1000,
					Value:     float64(reportID),
				},
			},
		}
		reportTss = append(reportTss, *ts)
	}
	if err := sink.metricsSink.SinkBatch(reportTss); err != nil {
		log.Error("batch insert failed", zap.Error(err))
		return err
	}
	return nil
}

// for cluster info in cache
// TODO(shenjun): use raw json message if we need support more fields
func (sink *ActiveQPSSink) getClusterInfoFromCache(key string) (name, region, deployType string, found bool) {
	value, ok := sink.clusterCache.HasGet(nil, ToUnsafeBytes(key))
	if !ok {
		return "", "", "", false
	}
	s := strings.Split(ToUnsafeString(value), ":")
	if len(s) != 3 {
		return "", "", "", false
	}
	return s[0], s[1], s[2], true
}

func (sink *ActiveQPSSink) setClusterInfoToCache(key, name, region, deployType string) {
	value := fmt.Sprintf("%s:%s:%s", name, region, deployType)
	sink.clusterCache.Set(ToUnsafeBytes(key), ToUnsafeBytes(value))
}

// for tenant info in cache
// TODO(shenjun): use raw json message if we need support more fields
func (sink *ActiveQPSSink) getTenantIDNameFromCache(key string) (id, name string, found bool) {
	value, ok := sink.tenantCache.HasGet(nil, ToUnsafeBytes(key))
	if !ok {
		return "", "", false
	}
	s := strings.Split(ToUnsafeString(value), ":")
	if len(s) != 2 {
		return "", "", false
	}
	return s[0], s[1], true
}

func (sink *ActiveQPSSink) setTenantIDNameToCache(key, id, name string) {
	value := fmt.Sprintf("%s:%s", id, name)
	sink.tenantCache.Set(ToUnsafeBytes(key), ToUnsafeBytes(value))
}

func (sink *ActiveQPSSink) getClusterMeta(uctx *session.CtxUser, param *InsertActiveQPSParam) (*clusterMeta, error) {
	meta := new(clusterMeta)
	// get cluster name
	{
		name, region, deployType, ok := sink.getClusterInfoFromCache(param.ClusterID)
		if !ok {
			// we have validated the org string is a number string
			orgID, _ := strconv.ParseUint(param.OrgID, 10, 64)
			org, err := sink.getDBaasOrgInfo(uctx, uint(orgID))
			if err != nil {
				log.Error("get dbaas org info failed", zap.String("org_id", param.OrgID), zap.Error(err))
				return nil, err
			}
			cluster, err := sink.getDBaasClusterInfo(uctx, param.ClusterID, org.ID)
			if err != nil {
				log.Error("get dbaas cluster info failed", zap.String("cluster_id", param.ClusterID), zap.Error(err))
				return nil, err
			}
			meta.clusterName = cluster.GetRawName()
			meta.region = cluster.Region
			meta.deployType = cluster.DeployType
			meta.tenantID = strconv.FormatUint(uint64(org.DbaasOrgID), 10)
			meta.tenantName = org.GetRawName()
			sink.setClusterInfoToCache(param.ClusterID, meta.clusterName, meta.region, meta.deployType)
			sink.setTenantIDNameToCache(param.OrgID, meta.tenantID, meta.tenantName)
		} else {
			meta.clusterName = name
			meta.region = region
			meta.deployType = deployType
		}
	}
	// get org name
	if len(meta.tenantID) == 0 {
		id, name, ok := sink.getTenantIDNameFromCache(param.OrgID)
		if !ok {
			// we have validated the org string is a number string
			orgID, _ := strconv.ParseUint(param.OrgID, 10, 64)
			org, err := sink.getDBaasOrgInfo(uctx, uint(orgID))
			log.Error("get dbaas org info failed", zap.String("org_id", param.OrgID), zap.Error(err))
			if err != nil {
				return nil, err
			}
			meta.tenantID = strconv.FormatUint(uint64(org.DbaasOrgID), 10)
			meta.tenantName = org.GetRawName()
			sink.setTenantIDNameToCache(param.OrgID, meta.tenantID, meta.tenantName)
		} else {
			meta.tenantID = id
			meta.tenantName = name
		}
	}
	return meta, nil
}

/* use BatchInsert instead
func (sink *ActiveQPSSink) InsertActiveOPSGinHandler(c *gin.Context) {
	param := new(InsertActiveQPSParam)
	if err := c.BindJSON(param); err != nil {
		if err != nil {
			_ = c.Error(error2.ErrBadRequest.New("param is invalid json"))
			return
		}
	}
	if err := param.Validate(); err != nil {
		_ = c.Error(error2.ErrBadRequest.New("%s", err.Error()))
		return
	}
	uctx := session.GetUser(c)
	if err := sink.InsertActiveQPS(uctx, param); err != nil {
		_ = c.Error(error2.ErrInternal.New("%s", err.Error()))
		return
	}
	c.JSON(http.StatusOK, common.SuccessResponse())
}
*/

func (sink *ActiveQPSSink) BatchInsertActiveOPSGinHandler(c *gin.Context) {
	param := BatchInsertActiveQPSParam{Items: []InsertActiveQPSParam{}}
	if err := c.BindJSON(&param); err != nil {
		if err != nil {
			log.Error("bind json failed", zap.Error(err))
			_ = c.Error(error2.ErrBadRequest.New("param is invalid json"))
			return
		}
	}
	if err := param.Validate(); err != nil {
		_ = c.Error(error2.ErrBadRequest.New("%s", err.Error()))
		return
	}
	uctx := session.GetUser(c)
	if err := sink.BatchInsertActiveQPS(uctx, param.Items); err != nil {
		_ = c.Error(error2.ErrInternal.New("%s", err.Error()))
		return
	}

	// ignore error about reports
	reportMap := insertReportsForQPS100(param.Items)
	_ = sink.BatchInsertActiveQPSRelatedReport(uctx, reportMap)

	c.JSON(http.StatusOK, common.SuccessResponse())
}

func insertReportsForQPS100(items []InsertActiveQPSParam) map[uint]*InsertActiveQPSParam {
	reportMap := make(map[uint]*InsertActiveQPSParam, 0)
	clinic, err := model.GetUserByUserName("clinic@pingcap.com")
	if err != nil {
		return nil
	}
	// insert report for cluster which qps >= 100
	for _, item := range items {
		orgID, err := strconv.Atoi(item.OrgID)
		if err != nil {
			continue
		}
		if item.Threshold < 10000 {
			continue
		}
		if item.EndUnix-item.StartUnix < 5*60 {
			continue
		}
		startUnix := item.StartUnix
		// reduce prometheus oom
		if item.EndUnix-item.StartUnix > 6*60*60 {
			startUnix = item.EndUnix - 6*60*60
		}
		arg, err := json.Marshal(map[string]any{
			"begin_ts": startUnix,
			"end_ts":   item.EndUnix,
		})
		r := db.Report{
			UUID:       uuid.NewString(),
			OrgID:      uint(orgID),
			ClusterID:  item.ClusterID,
			CreatorID:  clinic.ID,
			ReportType: "health2",
			Status:     string(db.StatusPending),
			Arg:        string(arg),
		}
		if err == nil {
			_ = model.InsertReport(&r)
		}
		id, err := model.ReportUUIDtoID(r.UUID)
		if err == nil {
			reportMap[id] = &item
		}
	}
	return reportMap
}

func (sink *ActiveQPSSink) getDBaasClusterInfo(uctx *session.CtxUser, clusterID string, orgID uint) (*db.Cluster, error) {
	if sink.getDBaasClusterFn == nil {
		return nil, errors.New("clusterInfoFetchFn is not initialized")
	}
	return sink.getDBaasClusterFn(uctx, clusterID, orgID)
}

func (sink *ActiveQPSSink) getDBaasOrgInfo(uctx *session.CtxUser, tenantID uint) (*db.Organization, error) {
	if sink.getDBaasOrgFn == nil {
		return nil, errors.New("getDBaasOrgFn is not initialized")
	}
	return sink.getDBaasOrgFn(uctx, tenantID)
}

func ToUnsafeString(b []byte) string {
	return *(*string)(unsafe.Pointer(&b))
}

func ToUnsafeBytes(s string) (b []byte) {
	return unsafe.Slice(unsafe.StringData(s), len(s))
}

func ToLittleEndianBytes(i uint64) []byte {
	key := make([]byte, 8)
	binary.LittleEndian.PutUint64(key, i)
	return key
}