package core

import (
	"context"
	"errors"
	"fmt"
	"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc"
	"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils/timeutils"
	"strconv"
	"sync"
	"time"

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

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

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

func (l *PageListTaskLogic) PageListTask(req *types.PageTaskReq) (resp *types.PageResult, err error) {
	limit := req.PageSize
	offset := req.PageSize * (req.PageNum - 1)
	resp = &types.PageResult{}
	var list []*types.TaskModel
	db := l.svcCtx.DbEngin.Model(&types.TaskModel{}).Table("task")

	db = db.Where("deleted_at is null")
	if req.Name != "" {
		db = db.Where("name LIKE ?", "%"+req.Name+"%")
	}

	//count total
	var total int64
	err = db.Count(&total).Error
	db.Limit(limit).Offset(offset)

	if err != nil {
		return resp, err
	}
	err = db.Order("created_time desc").Find(&list).Error
	if err != nil {
		return nil, result.NewDefaultError(err.Error())
	}

	// 更新智算任务状态
	chs := [2]chan struct{}{make(chan struct{}), make(chan struct{})}
	go l.updateTaskStatus(list, chs[0])
	go l.updateAiTaskStatus(list, chs[1])

	for _, model := range list {
		if model.StartTime != "" && model.EndTime == "" {
			startTime := timeutils.TimeStringToGoTime(model.StartTime)
			model.RunningTime = int64(time.Now().Sub(startTime).Seconds())
		}
		if model.StartTime != "" && model.EndTime != "" {
			startTime := timeutils.TimeStringToGoTime(model.StartTime)
			endTime := timeutils.TimeStringToGoTime(model.EndTime)
			model.RunningTime = int64(endTime.Sub(startTime).Seconds())
		}
	}
	resp.List = &list
	resp.PageSize = req.PageSize
	resp.PageNum = req.PageNum
	resp.Total = total

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

func (l *PageListTaskLogic) updateTaskStatus(tasklist []*types.TaskModel, ch chan<- struct{}) {
	list := make([]*types.TaskModel, len(tasklist))
	copy(list, tasklist)
	for i := len(list) - 1; i >= 0; i-- {
		if list[i].AdapterTypeDict != 1 || list[i].Status == constants.Succeeded || list[i].Status == constants.Failed {
			list = append(list[:i], list[i+1:]...)
		}
	}

	if len(list) == 0 {
		ch <- struct{}{}
		return
	}

	task := list[0]
	for i := range list {
		earliest, _ := time.Parse(time.RFC3339, task.UpdatedTime)
		latest, _ := time.Parse(time.RFC3339, list[i].UpdatedTime)
		if latest.Before(earliest) {
			task = list[i]
		}
	}

	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())
		ch <- struct{}{}
		return
	}

	if len(aiTask) == 0 {
		tx = l.svcCtx.DbEngin.Model(task).Table("task").Where("deleted_at is null").Updates(task)
		if tx.Error != nil {
			logx.Errorf(tx.Error.Error())
			ch <- struct{}{}
			return
		}
		ch <- struct{}{}
		return
	}

	if len(aiTask) == 1 {
		if aiTask[0].Status == constants.Completed {
			task.Status = constants.Succeeded
		} else {
			task.Status = aiTask[0].Status
		}
		task.StartTime = aiTask[0].StartTime
		task.EndTime = aiTask[0].EndTime
		task.UpdatedTime = time.Now().Format(constants.Layout)
		tx = l.svcCtx.DbEngin.Model(task).Table("task").Where("deleted_at is null").Updates(task)
		if tx.Error != nil {
			logx.Errorf(tx.Error.Error())
			ch <- struct{}{}
			return
		}
		ch <- struct{}{}
		return
	}

	for i := len(aiTask) - 1; i >= 0; i-- {
		if aiTask[i].StartTime == "" {
			task.Status = aiTask[i].Status
			aiTask = append(aiTask[:i], aiTask[i+1:]...)
		}
	}

	if len(aiTask) == 0 {
		task.UpdatedTime = time.Now().Format(constants.Layout)
		tx = l.svcCtx.DbEngin.Table("task").Model(task).Updates(task)
		if tx.Error != nil {
			logx.Errorf(tx.Error.Error())
			ch <- struct{}{}
			return
		}
		ch <- struct{}{}
		return
	}

	start, _ := time.ParseInLocation(constants.Layout, aiTask[0].StartTime, time.Local)
	end, _ := time.ParseInLocation(constants.Layout, aiTask[0].EndTime, time.Local)

	var status string
	var count int
	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.Pending {
			status = a.Status
			continue
		}

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

		if a.Status == constants.Completed {
			count++
			continue
		}
	}

	if count == len(aiTask) {
		status = constants.Succeeded
	}

	if status != "" {
		task.Status = status
		task.StartTime = start.Format(constants.Layout)
		task.EndTime = end.Format(constants.Layout)
	}

	task.UpdatedTime = time.Now().Format(constants.Layout)
	tx = l.svcCtx.DbEngin.Table("task").Model(task).Updates(task)
	if tx.Error != nil {
		logx.Errorf(tx.Error.Error())
		ch <- struct{}{}
		return
	}
	ch <- struct{}{}
}

func (l *PageListTaskLogic) updateAiTaskStatus(tasklist []*types.TaskModel, ch chan<- struct{}) {
	list := make([]*types.TaskModel, len(tasklist))
	copy(list, tasklist)
	for i := len(list) - 1; i >= 0; i-- {
		if list[i].AdapterTypeDict != 1 || list[i].Status == constants.Succeeded || list[i].Status == constants.Failed {
			list = append(list[:i], list[i+1:]...)
		}
	}

	if len(list) == 0 {
		ch <- struct{}{}
		return
	}

	task := list[0]
	for i := range list {
		earliest, _ := time.Parse(constants.Layout, task.UpdatedTime)
		latest, _ := time.Parse(constants.Layout, list[i].UpdatedTime)
		if latest.Before(earliest) {
			task = list[i]
		}
	}

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

	if len(aiTaskList) == 0 {
		ch <- struct{}{}
		return
	}

	var wg sync.WaitGroup
	for _, aitask := range aiTaskList {
		t := aitask
		if t.Status == constants.Completed || t.Status == constants.Failed {
			continue
		}
		wg.Add(1)
		go func() {
			trainingTask, err := l.svcCtx.Scheduler.AiService.AiCollectorAdapterMap[strconv.FormatInt(t.AdapterId, 10)][strconv.FormatInt(t.ClusterId, 10)].GetTrainingTask(l.ctx, t.JobId)
			if err != nil {
				msg := fmt.Sprintf("AiTaskId: %v, clusterId: %v , JobId: %v,  error: %v \n", t.Id, t.ClusterId, t.JobId, err.Error())
				logx.Errorf(errors.New(msg).Error())
				wg.Done()
				return
			}
			if trainingTask == nil {
				wg.Done()
				return
			}
			t.Status = trainingTask.Status
			t.StartTime = trainingTask.Start
			t.EndTime = trainingTask.End
			err = l.svcCtx.Scheduler.AiStorages.UpdateAiTask(t)
			if err != nil {
				msg := fmt.Sprintf("AiTaskId: %v, clusterId: %v , JobId: %v,  error: %v \n", t.Id, t.ClusterId, t.JobId, err.Error())
				logx.Errorf(errors.New(msg).Error())
				wg.Done()
				return
			}
			wg.Done()
		}()
	}
	wg.Wait()
	ch <- struct{}{}
}
