aiforge/services/reward/cloubrain_deduct.go

package reward

import (
	"code.gitea.io/gitea/models"
	"code.gitea.io/gitea/modules/log"
	"code.gitea.io/gitea/modules/setting"
	"fmt"
	"time"
)

var (
	ResourceSpecs      *models.ResourceSpecs
	TrainResourceSpecs *models.ResourceSpecs
)

const RUN_CLOUDBRAIN_TASK_TITTLE = "运行云脑任务"

func StartAndGetCloudBrainPointDeductTask(task models.Cloudbrain) (*models.RewardPeriodicTask, error) {
	if !setting.CloudBrainPaySwitch {
		return nil, nil
	}

	unitPrice := models.GetCloudbrainTaskUnitPrice(task)
	if unitPrice == 0 {
		log.Debug("finish StartAndGetCloudBrainPointDeductTask,  UnitPrice = 0 task.ID=%d", task.ID)
		return nil, nil
	}

	return StartAndGetPeriodicTask(&models.StartPeriodicTaskOpts{
		SourceType:   models.SourceTypeRunCloudbrainTask,
		SourceId:     getCloudBrainPointTaskSourceId(task),
		TargetUserId: task.UserID,
		RequestId:    getCloudBrainPointTaskSourceId(task),
		OperateType:  models.OperateTypeDecrease,
		Delay:        setting.CloudBrainPayDelay,
		Interval:     setting.CloudBrainPayInterval,
		UnitAmount:   unitPrice,
		RewardType:   models.RewardTypePoint,
		StartTime:    time.Unix(int64(task.StartTime), 0),
		Tittle:       RUN_CLOUDBRAIN_TASK_TITTLE,
	})
}

func StopCloudBrainPointDeductTask(task models.Cloudbrain) {
	StopPeriodicTask(models.SourceTypeRunCloudbrainTask, getCloudBrainPointTaskSourceId(task), models.OperateTypeDecrease)
}

func getCloudBrainPointTaskSourceId(task models.Cloudbrain) string {
	return fmt.Sprint(task.ID)
}

var firstTimeFlag = true

func StartCloudbrainPointDeductTask() {
	defer func() {
		if err := recover(); err != nil {
			combinedErr := fmt.Errorf("%s\n%s", err, log.Stack(2))
			log.Error("PANIC:%v", combinedErr)
		}
	}()
	log.Debug("try to run CloudbrainPointDeductTask")
	end := time.Now()
	start := end.Add(-30 * time.Minute)
	if firstTimeFlag {
		//When it is executed for the first time, it needs to process the tasks of the last 1 hours.
		//This is done to prevent the application from hanging for a long time
		start = end.Add(-3 * time.Hour)
		firstTimeFlag = false
	}
	taskList, err := models.GetStartedCloudbrainTaskByUpdatedUnix(start, end)
	if err != nil {
		log.Error("GetStartedCloudbrainTaskByUpdatedUnix error. %v", err)
		return
	}
	if taskList == nil || len(taskList) == 0 {
		log.Debug("No cloudbrain task need handled")
		return
	}
	for _, t := range taskList {
		DeductPoint4Cloudbrain(t, end)
	}
}

func DeductPoint4Cloudbrain(t models.Cloudbrain, now time.Time) error {

	if t.StartTime == 0 {
		return nil
	}

	task, err := StartAndGetCloudBrainPointDeductTask(t)
	if err != nil {
		log.Error("run cloudbrain point deduct task error,err=%v", err)
		return err
	}
	if task == nil {
		return nil
	}
	if task.Status == models.PeriodicTaskStatusFinished {
		log.Info("Periodic task is finished")
		return nil
	}

	if t.EndTime > 0 {
		endTime := time.Unix(int64(t.EndTime), 0)
		RunRewardTask(*task, endTime)
		models.StopPeriodicTask(task.ID, task.OperateSerialNo, endTime)
	} else {
		RunRewardTask(*task, now)
	}
	return nil
}