package service

import (
	"encoding/json"
	"fmt"
	"git.quantgroup.cn/DevOps/enoch/service/conf"
	"github.com/gomodule/redigo/redis"
	"github.com/influxdata/influxdb/client/v2"
	"log"
	"strconv"
	"time"
)

var httpMethod = map[string]string{
	"get": "", "GET": "", "post": "", "options": "", "put": "", "delete": "", "head": "",
}

func MsgProcess(msg string) {
	traceMsg := make([]TraceMsg, 3) //[]TraceMsg{}
	err := json.Unmarshal([]byte(msg), &traceMsg)
	if err != nil {
		fmt.Println(err)
	}

	//msgRedisProcess(traceMsg)
	msgInfluxProcess(traceMsg)
}

var batchSize = 5000
var pointSlice = make([]*client.Point, 0, batchSize)

func msgInfluxProcess(traceMsgs []TraceMsg) {

	for _, traceMsg := range traceMsgs {
		if traceMsg.Kind != "SERVER" {
			break
		}
		path := traceMsg.Name

		if _, ok := httpMethod[path]; ok {
			path = traceMsg.Tags.HttpMethod + " " + traceMsg.Tags.HttpPath
		}

		sysName := traceMsg.LocalEndpoint.ServiceName

		fields := make(map[string]interface{})
		fields["duration"] = traceMsg.Duration / 1000
		fields["traceId"] = traceMsg.TraceId
		tags := make(map[string]string, )
		tags["sys_name"] = sysName
		tags["path"] = path
		tags["host"] = traceMsg.LocalEndpoint.Ipv4
		tags["kind"] = traceMsg.Kind
		tags["duration_tag"] = strconv.Itoa(traceMsg.Duration)

		bytes, err := json.Marshal(traceMsg)

		msg := string(bytes)

		tags["msg"] = msg

		if err != nil {
			log.Fatal(err)
		}

		unix := time.Unix(0, traceMsg.Timestamp*1000)

		if len(pointSlice) == batchSize {
			go batchWrite(pointSlice)
			pointSlice = make([]*client.Point, 0, batchSize)
		}

		point, _ := client.NewPoint("trace_info", tags, fields, unix)
		pointSlice = append(pointSlice, point)

		if err != nil {
			log.Fatal(err, msg)
		}
	}

}

func batchWrite(pointArray []*client.Point) {
	c := conf.NewClient()
	defer func() { _ = c.Close() }()

	points, err := client.NewBatchPoints(client.BatchPointsConfig{
		Database: "monitor",
	})
	if err != nil {
		log.Fatal(err)
	}

	points.AddPoints(pointArray)
	err = c.Write(points)
	fmt.Println("写入数据",len(pointArray))
	if err != nil {
		log.Fatal(err)
	}

}

func msgRedisProcess(traceMsg []TraceMsg) {
	conn := conf.Pool.Get()
	defer func() { _ = conn.Close() }()
	for _, v := range traceMsg {
		if v.Kind != "SERVER" {
			break
		}
		//处理消息

		path := v.Name

		if _, ok := httpMethod[path]; ok {
			path = v.Tags.HttpMethod + " " + v.Tags.HttpPath
		}

		var counterKey = keyGenerator("counter", v.LocalEndpoint.ServiceName)

		var field = path
		//1. 接口计数器
		_, err := conn.Do("zincrby", counterKey, 1, field)
		if err != nil {
			fmt.Println("执行出错: ", err)
		}
		//2. 接口耗时排行
		var durationKey = keyGenerator("duration", v.LocalEndpoint.ServiceName)

		var member = path

		//v.TraceId

		casScoreAdd(conn, durationKey, member, v.Duration)

	}
}

func keyGenerator(prefix string, info string) string {
	day := time.Now().Format("20060102")
	return prefix + ":" + info + ":" + day
}

func casScoreAdd(conn redis.Conn, durationKey string, member string, duration int) {
	durationInfo, err := conn.Do("zscore", durationKey, member)
	var oldDuration = duration
	if durationInfo != nil {
		s := string(durationInfo.([]uint8))
		i, err := strconv.Atoi(s)
		if err != nil {
			fmt.Println("转换durationInfo 出错", err)
		}
		if i >= duration {
			//如果大于当前值, 中断
			return
		}
		oldDuration = i
	}
	err = conn.Send("watch", durationKey)
	err = conn.Send("multi")
	err = conn.Send("zadd", durationKey, duration, member)
	reply, err := conn.Do("exec")
	if err != nil {
		fmt.Println("执行出错 : ", err)
	}
	if reply != nil {
		return
	}
	//loop
	casScoreAdd(conn, durationKey, member, oldDuration)
}

func Duration(day string, fun func(sysName string, durations map[string]string)) {

	conn := conf.Pool.Get()
	defer func() { _ = conn.Close() }()
	matchPattern := "duration*" + day
	reply, err := redis.Values(conn.Do("scan", 0, "match", matchPattern, "count", 10000))
	if err != nil {
		fmt.Print("Scan 扫描key出错", err)
	}

	redisKeys := reply[1].([]interface{})
	for _, redisKey := range redisKeys {

		reply2, err := redis.StringMap(conn.Do("zrevrangebyscore", redisKey, "+inf", 200000, "withscores"))
		if err != nil {
			fmt.Print("获取出错了", err)
		}
		fun(string(redisKey.([]uint8)), reply2)
	}

}

func Counter(day string, fun func(sysName string, durations map[string]string)) {
	conn := conf.Pool.Get()
	defer func() { _ = conn.Close() }()
	matchPattern := "counter*" + day
	reply, err := redis.Values(conn.Do("scan", 0, "match", matchPattern, "count", 10000))
	if err != nil {
		fmt.Print("Scan 扫描key出错", err)
	}
	redisKeys := reply[1].([]interface{})
	for _, redisKey := range redisKeys {

		reply2, err := redis.StringMap(conn.Do("zrevrange", redisKey, 0, 300, "withscores"))
		if err != nil {
			fmt.Print("获取出错了", err)
		}
		fun(string(redisKey.([]uint8)), reply2)
	}

}
