/*

 Copyright (c) [2023] [pcm]
 [pcm-coordinator] is licensed under Mulan PSL v2.
 You can use this software according to the terms and conditions of the Mulan PSL v2.
 You may obtain a copy of Mulan PSL v2 at:
          http://license.coscl.org.cn/MulanPSL2
 THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 EITHER EXPaRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details.

*/

package core

import (
	"context"
	"fmt"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
	"strconv"
	"time"

	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/types"

	"github.com/zeromicro/go-zero/core/logx"
)

type TaskListLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewTaskListLogic(ctx context.Context, svcCtx *svc.ServiceContext) *TaskListLogic {
	return &TaskListLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *TaskListLogic) TaskList(req *types.TaskListReq) (resp *types.TaskListResp, err error) {
	resp = &types.TaskListResp{}

	// 查询任务数据
	var tasks []models.Task
	offset := (req.PageNum - 1) * req.PageSize
	tx := l.svcCtx.DbEngin.Order("created_time desc").Offset(offset).Limit(req.PageSize).Find(&tasks)
	if tx.Error != nil {
		logx.Error(err)
		return nil, tx.Error
	}
	if len(tasks) == 0 {
		return nil, nil
	}

	// 更新智算任务状态
	var ch = make(chan struct{})
	//go l.updateAitaskStatus(tasks, ch)

	// 查询任务总数
	l.svcCtx.DbEngin.Model(&models.Task{}).Count(&resp.TotalCount)

	// 查询任务异常数
	l.svcCtx.DbEngin.Model(&models.Task{}).Where("status = ?", "Failed").Count(&resp.AlarmCount)

	// 正常任务数
	resp.NormalCount = resp.TotalCount - resp.AlarmCount
	for _, task := range tasks {
		type PInfo struct {
			Id      int64
			Name    string
			Address string
		}
		pInfo := PInfo{}
		tx := l.svcCtx.DbEngin.Raw("SELECT id,name,address from sc_participant_phy_info  where id  in (SELECT CONCAT_WS(',',GROUP_CONCAT(DISTINCT h.participant_id) ,GROUP_CONCAT(DISTINCT  a.participant_id) ,GROUP_CONCAT(DISTINCT c.participant_id),GROUP_CONCAT(DISTINCT  v.participant_id))as service_name from task t left join hpc h on t.id  = h.task_id left join cloud c  on t.id  = c.task_id left join ai a on t.id  = a.task_id left join vm v on t.id  = v.task_id  where t.id  = ?)", task.Id).Scan(&pInfo)
		if tx.Error != nil {
			logx.Error(err)
			return nil, tx.Error
		}
		// 获取p端状态
		pStatus := "Unknown"
		key := fmt.Sprintf("ps_%s-%d", pInfo.Address, pInfo.Id)
		hGetAll := l.svcCtx.RedisClient.HGetAll(l.ctx, key)
		if hGetAll.Err() != nil {
			return nil, err
		}
		if len(hGetAll.Val()) != 0 {
			parseInt, err := strconv.ParseInt(hGetAll.Val()["lastHeartbeat"], 10, 64)
			if err != nil {
				return nil, err
			}

			if (time.Now().Unix() - parseInt) < 15 {

				pStatus = "Normal"
			}
		}
		resp.Tasks = append(resp.Tasks, types.Task{
			Id:                task.Id,
			Name:              task.Name,
			Status:            task.Status,
			StartTime:         task.StartTime.Format("2006-01-02 15:04:05"),
			EndTime:           task.EndTime.Format("2006-01-02 15:04:05"),
			ParticipantId:     pInfo.Id,
			ParticipantName:   pInfo.Name,
			ParticipantStatus: pStatus,
		})

	}

	select {
	case _ = <-ch:
		return resp, nil
	case <-time.After(1 * time.Second):
		return resp, nil
	}
}

func (l *TaskListLogic) updateAitaskStatus(tasks []models.Task, ch chan<- struct{}) {
	for _, task := range tasks {
		if task.AdapterTypeDict != "1" {
			continue
		}
		if task.Status == constants.Succeeded {
			continue
		}

		var aiTask []*models.TaskAi
		tx := l.svcCtx.DbEngin.Raw("select * from task_ai where `task_id` = ? ", task.Id).Scan(&aiTask)
		if tx.Error != nil {
			logx.Errorf(tx.Error.Error())
			return
		}

		start, _ := time.ParseInLocation(constants.Layout, aiTask[0].StartTime, time.Local)
		end, _ := time.ParseInLocation(constants.Layout, aiTask[0].EndTime, time.Local)
		var status = constants.Succeeded
		for _, a := range aiTask {
			s, _ := time.ParseInLocation(constants.Layout, a.StartTime, time.Local)
			e, _ := time.ParseInLocation(constants.Layout, a.EndTime, time.Local)

			if s.Before(start) {
				start = s
			}

			if e.After(end) {
				end = e
			}

			if a.Status == constants.Failed {
				status = a.Status
				break
			}

			if a.Status == constants.Running {
				status = a.Status
				continue
			}
		}

		task.Status = status
		task.StartTime = &start
		task.EndTime = &end

		tx = l.svcCtx.DbEngin.Updates(task)
		if tx.Error != nil {
			return
		}
	}
	ch <- struct{}{}
}
