package cronjob

import (
	"ai_developer_admin/libs/postgres"
	"ai_developer_admin/libs/redis"
	"ai_developer_admin/libs/registry"
	"ai_developer_admin/libs/snowflake"
	"ai_developer_admin/models"
	"context"
	"encoding/json"
	"fmt"
	"github.com/beego/beego/v2/core/logs"
	beego "github.com/beego/beego/v2/server/web"
	qdb "github.com/questdb/go-questdb-client/v2"
	"github.com/robfig/cron/v3"
	"sort"
	"strconv"
	"time"
)

var loopCronTask = cron.New(cron.WithSeconds())
var debitTask = cron.New(cron.WithSeconds())
var registryTask = cron.New(cron.WithSeconds())

var HeatKey = "task:heat"

var format = "2006-01-02T15:04:05.000000Z"

func Start() {
	//defer loopCronTask.Stop()
	startHeatKey()
	startDebit()
	//startRegistBackend()
}

func startHeatKey() {
	//spec := "0 0 * * * ?" //"@every 1h"
	spec := "@every 1h" //"@every 1h"
	//spec := "*/1 * * * * ?" //"@every 1h"
	loopCronTask.AddFunc(spec, func() {
		logs.Debug("loopCronTask")
		sql := "SELECT type,count(type) FROM bills GROUP BY type ORDER BY count DESC;"
		tasks, err := postgres.CountTasks(sql)
		if err != nil {
			return
		}
		dataString, err := redis.GetDataToString(HeatKey)
		var response []*models.TaskHeat
		if err != nil {
			return
		}
		err = json.Unmarshal([]byte(dataString), &response)
		if err != nil {
			return
		}
		for _, value := range tasks {
			taskId, err := strconv.Atoi(value.Type)
			if err != nil {
				continue
			}
			count, _ := strconv.Atoi(value.Count)
			for _, taskType := range response {
				if taskType.TaskId == taskId {
					taskType.Count = int64(count)
				}
			}
		}
		//for _, value := range tasks {
		//	taskId, err := strconv.Atoi(value.Type)
		//	if err != nil {
		//		continue
		//	}
		//	taskType, err1 := odysseus.GetTaskType(int64(taskId))
		//	if err1 != nil {
		//		continue
		//	}
		//	var hardwareRequire models.Hardware
		//	eer := json.Unmarshal([]byte(taskType.HardwareRequire), &hardwareRequire)
		//	if eer != nil {
		//
		//	}
		//	count, _ := strconv.Atoi(value.Count)
		//	retask := models.TaskHeat{
		//		TaskId:          taskId,
		//		User:            taskType.Username,
		//		Pwd:             taskType.Password,
		//		Repository:      taskType.ImageUrl,
		//		SignUrl:         taskType.SignUrl,
		//		ImageName:       taskType.ImageName,
		//		ImageId:         taskType.ImageId,
		//		HardwareRequire: hardwareRequire,
		//		Count:           int64(count),
		//	}
		//	response = append(response, retask)
		//}
		sort.Slice(response, func(i, j int) bool {
			return response[i].Count > response[j].Count
		})
		data, err := json.Marshal(response)
		if err == nil {
			redis.SetKeyAndData(HeatKey, string(data), 0)
		}
	})
	loopCronTask.Start()
}

func startDebit() {
	logs.Debug("startDebit")
	//spec := "*/50 */23 * * * ?" //"@every 1h"
	//spec := "01 01 00 * * ?"
	spec, _ := beego.AppConfig.String("debitTime")
	//spec := "@every 1m"
	dbhost, _ := beego.AppConfig.String("postgreshost")
	dbport, _ := beego.AppConfig.Int("senderport")

	questAddr := fmt.Sprintf("%s:%d", dbhost, dbport)
	debitTask.AddFunc(spec, func() {
		logs.Debug("startDebit debitTask")
		ctx := context.TODO()
		addrOpt := qdb.WithAddress(questAddr)
		sender, err := qdb.NewLineSender(ctx, addrOpt)

		if err != nil {
			logs.Debug("startDebit NewLineSender = %s", err.Error())
			return
		}
		defer sender.Close()
		// Make sure to close the sender on exit to release resources.

		currentTime := time.Now()
		temp := fmt.Sprintf("-%dh", 2)
		m, _ := time.ParseDuration(temp)
		dayTime := currentTime.Add(m)
		endTime := time.Date(dayTime.Year(), dayTime.Month(), dayTime.Day(), 23, 59, 59, 0, dayTime.Location())
		startTime := time.Date(dayTime.Year(), dayTime.Month(), dayTime.Day(), 0, 0, 0, 0, dayTime.Location())
		start := fmt.Sprintf(startTime.Format(format))
		end := fmt.Sprintf(endTime.Format(format))

		logs.Debug("startDebit before sql = %s", dayTime)

		sql := fmt.Sprintf("SELECT time, uid,sum(fee) AS amount FROM bills WHERE uid != '0' and time >= '%s' and time <= '%s' SAMPLE BY 1d ALIGN TO CALENDAR GROUP BY uid,time;", start, end)
		data, err := postgres.CountFunds(sql)
		if err != nil {
			logs.Debug("startDebit postgres CountFunds = %s", err.Error())
			return
		}
		if data == nil {
			logs.Debug("startDebit data = nil")
			return
		}

		//sql = "SELECT Max(id) AS count FROM funds;"
		//max, err := postgres.QueryTotal(sql)
		//if err != nil {
		//	return
		//}

		for _, fund := range data {
			//tradeTime, _ := time.Parse(fund.Time, format)
			id, _ := snowflake.NextId()

			uid, _ := strconv.Atoi(fund.Uid)
			amount, _ := strconv.Atoi(fund.Amount)

			nanoseconds := int64(uint64(dayTime.UnixNano()))
			seconds := nanoseconds / 1e9

			order, _ := snowflake.NextId()
			orderId := strconv.FormatInt(int64(order), 10)

			err = sender.Table("funds").
				Symbol("order_id", orderId).
				Symbol("remark", "").
				Symbol("channel_serial", "").
				Int64Column("uid", int64(uid)).
				Int64Column("amount", int64(amount)).
				Int64Column("balance", 0).
				Int64Column("trade_channel", 0).
				Int64Column("status", 5).
				Int64Column("id", int64(id)).
				TimestampColumn("trade_time", time.Unix(seconds, nanoseconds%1e9)).
				Int64Column("trade_flow", int64(models.Expenditure)).
				Int64Column("trade_type", int64(models.Spending)).
				AtNow(ctx)

			if err != nil {
				logs.Debug("startDebit sender = %s,id = %d", err.Error(), id)
			}
			//break
		}

		err = sender.Flush(ctx)
		if err != nil {
			logs.Debug("startDebit Flush = %s", err.Error())
		}
	})
	debitTask.Start()
}

func startRegistBackend() {
	spec := "@every 1m"
	registryTask.AddFunc(spec, func() {
		logs.Debug("startRegistBackend")
		registry.RegistryBackend()
	})
	debitTask.Start()
}

//func Start() {
//	//defer loopCronTask.Stop()
//
//	spec := "*/1 * * * * ?" //"@every 1h"
//	loopCronTask.AddFunc(spec, func() {
//		logs.Debug("loopCronTask")
//		//sql := "SELECT type,count(type) FROM bills GROUP BY type ORDER BY count DESC;"
//		//tasks, err := postgres.CountTasks(sql)
//		//if err != nil {
//		//	return
//		//}
//		var response []models.TaskHeat
//		//for _, value := range tasks {
//		//	taskId, err := strconv.Atoi(value.Type)
//		//	if err != nil {
//		//		continue
//		//	}
//		taskType, err1 := odysseus.GetTaskType(int64(12))
//		if err1 != nil {
//			//continue
//		}
//		var hardwareRequire models.Hardware
//		eer := json.Unmarshal([]byte(taskType.HardwareRequire), &hardwareRequire)
//		if eer != nil {
//
//		}
//		//count, _ := strconv.Atoi(value.Count)
//		retask := models.TaskHeat{
//			TaskId:          12,
//			User:            taskType.Username,
//			Pwd:             taskType.Password,
//			Repository:      taskType.ImageUrl,
//			SignUrl:         taskType.SignUrl,
//			ImageName:       taskType.ImageName,
//			ImageId:         taskType.ImageId,
//			HardwareRequire: hardwareRequire,
//			Count:           int64(0),
//		}
//		response = append(response, retask)
//		//}
//		data, err := json.Marshal(response)
//		if err == nil {
//			redis.SetKeyAndData("task:heat", string(data), 0)
//		}
//	})
//	loopCronTask.Start()
//}
