aiforge/services/reward/operator.go

package reward

import (
	"code.gitea.io/gitea/models"
	"code.gitea.io/gitea/modules/log"
	"code.gitea.io/gitea/modules/redis/redis_key"
	"code.gitea.io/gitea/modules/redis/redis_lock"
	"code.gitea.io/gitea/services/reward/point"
	"errors"
	"fmt"
	"time"
)

var RewardOperatorMap = map[string]RewardOperator{
	fmt.Sprint(models.RewardTypePoint): new(point.PointOperator),
}

type RewardOperator interface {
	IsLimited(ctx *models.RewardOperateContext) error
	Operate(ctx *models.RewardOperateContext) error
}

func Operate(ctx *models.RewardOperateContext) error {
	defer func() {
		if err := recover(); err != nil {
			combinedErr := fmt.Errorf("%s\n%s", err, log.Stack(2))
			log.Error("PANIC:%v", combinedErr)
		}
	}()
	if !checkRewardOperationParam(ctx) {
		log.Error("send reward error,param incorrect")
		return errors.New("param incorrect")
	}
	//add lock
	var rewardLock = redis_lock.NewDistributeLock(redis_key.RewardOperateLock(ctx.RequestId, ctx.SourceType.Name(), ctx.OperateType.Name()))
	isOk, err := rewardLock.Lock(3 * time.Second)
	if err != nil {
		return err
	}
	if !isOk {
		log.Info("duplicated reward request,targetUserId=%d requestId=%s", ctx.TargetUserId, ctx.RequestId)
		return nil
	}
	defer rewardLock.UnLock()

	//is handled before?
	isHandled, err := isHandled(ctx.SourceType.Name(), ctx.RequestId, ctx.OperateType.Name())
	if err != nil {
		log.Error("reward is handled error,%v", err)
		return err
	}
	if isHandled {
		log.Info("reward has been handled,ctx=%+v", ctx)
		return nil
	}

	//get operator
	operator := GetOperator(ctx.Reward.Type)
	if operator == nil {
		return errors.New("operator of reward type is not exist")
	}

	if ctx.OperateType == models.OperateTypeIncrease {
		//is limited?
		if err := operator.IsLimited(ctx); err != nil {
			log.Info("operator IsLimited, err=%v", err)
			return err
		}
	}

	//new reward operate record
	recordId, err := initRewardOperateRecord(ctx)
	if err != nil {
		return err
	}

	ctx.SourceId = recordId

	//operate
	if err := operator.Operate(ctx); err != nil {
		UpdateRewardRecordToFinalStatus(ctx.SourceType.Name(), ctx.RequestId, models.OperateStatusFailed)
		return err
	}

	UpdateRewardRecordToFinalStatus(ctx.SourceType.Name(), ctx.RequestId, models.OperateStatusSucceeded)
	NotifyRewardOperation(ctx.TargetUserId, ctx.Reward.Amount, ctx.Reward.Type, ctx.OperateType)
	return nil
}

func checkRewardOperationParam(ctx *models.RewardOperateContext) bool {
	if ctx.Reward.Type == "" {
		return false
	}
	return true
}

func GetOperator(rewardType models.RewardType) RewardOperator {
	return RewardOperatorMap[rewardType.Name()]
}

func isHandled(sourceType string, requestId string, operateType string) (bool, error) {
	_, err := models.GetPointOperateRecordBySourceTypeAndRequestId(sourceType, requestId, operateType)
	if err != nil {
		if models.IsErrRecordNotExist(err) {
			return false, nil
		}
		return false, err
	}
	return true, nil

}

func initRewardOperateRecord(ctx *models.RewardOperateContext) (string, error) {
	sn, err := generateOperateSerialNo(ctx.OperateType, ctx.Reward.Type)
	if err != nil {
		return "", err
	}
	record := &models.RewardOperateRecord{
		UserId:      ctx.TargetUserId,
		Amount:      ctx.Reward.Amount,
		RewardType:  ctx.Reward.Type.Name(),
		SourceType:  ctx.SourceType.Name(),
		SourceId:    ctx.SourceId,
		RequestId:   ctx.RequestId,
		OperateType: ctx.OperateType.Name(),
		Status:      models.OperateStatusOperating,
		Remark:      ctx.Remark,
		Tittle:      ctx.Tittle,
		SerialNo:    sn,
	}
	_, err = models.InsertRewardOperateRecord(record)
	if err != nil {
		return "", err
	}
	return record.SerialNo, nil
}

func createPeriodicRewardOperateRecord(ctx *models.StartPeriodicTaskOpts) (string, error) {
	sn, err := generateOperateSerialNo(ctx.OperateType, ctx.RewardType)
	if err != nil {
		return "", err
	}
	record := &models.RewardOperateRecord{
		UserId:      ctx.TargetUserId,
		Amount:      0,
		RewardType:  ctx.RewardType.Name(),
		SourceType:  ctx.SourceType.Name(),
		SourceId:    ctx.SourceId,
		RequestId:   ctx.RequestId,
		OperateType: ctx.OperateType.Name(),
		Status:      models.OperateStatusOperating,
		Remark:      ctx.Remark,
		Tittle:      ctx.Tittle,
		SerialNo:    sn,
	}
	_, err = models.InsertRewardOperateRecord(record)
	if err != nil {
		return "", err
	}
	return record.SerialNo, nil
}

func UpdateRewardRecordToFinalStatus(sourceType, requestId, newStatus string) error {
	_, err := models.UpdateRewardRecordToFinalStatus(sourceType, requestId, newStatus)
	if err != nil {
		return err
	}
	return nil
}

func StartPeriodicTaskAsyn(opts *models.StartPeriodicTaskOpts) {
	go StartAndGetPeriodicTask(opts)
}

func StartAndGetPeriodicTask(opts *models.StartPeriodicTaskOpts) (*models.RewardPeriodicTask, error) {
	defer func() {
		if err := recover(); err != nil {
			combinedErr := fmt.Errorf("%s\n%s", err, log.Stack(2))
			log.Error("PANIC:%v", combinedErr)
		}
	}()
	//add lock
	var rewardLock = redis_lock.NewDistributeLock(redis_key.RewardOperateLock(opts.RequestId, opts.SourceType.Name(), opts.OperateType.Name()))
	isOk, err := rewardLock.Lock(3 * time.Second)
	if err != nil {
		return nil, err
	}
	if !isOk {
		log.Info("duplicated operate request,targetUserId=%d requestId=%s", opts.TargetUserId, opts.RequestId)
		return nil, nil
	}
	defer rewardLock.UnLock()

	_, err = models.GetPointOperateRecordBySourceTypeAndRequestId(opts.SourceType.Name(), opts.RequestId, opts.OperateType.Name())
	if err == nil {
		task, err := models.GetPeriodicTaskBySourceIdAndType(opts.SourceType, opts.SourceId, opts.OperateType)
		if err != nil {
			log.Error("GetPeriodicTaskBySourceIdAndType error,%v", err)
			return nil, err
		}
		return task, nil
	}

	if err != nil && !models.IsErrRecordNotExist(err) {
		log.Error("operate is handled error,%v", err)
		return nil, err
	}

	//new reward operate record
	recordId, err := createPeriodicRewardOperateRecord(opts)
	if err != nil {
		return nil, err
	}

	if err = NewRewardPeriodicTask(recordId, opts); err != nil {
		UpdateRewardRecordToFinalStatus(opts.SourceType.Name(), opts.RequestId, models.OperateStatusFailed)
		return nil, err
	}

	task, err := models.GetPeriodicTaskBySourceIdAndType(opts.SourceType, opts.SourceId, opts.OperateType)
	if err != nil {
		log.Error("GetPeriodicTaskBySourceIdAndType error,%v", err)
		return nil, err
	}
	return task, nil
}

func StopPeriodicTaskAsyn(sourceType models.SourceType, sourceId string, operateType models.RewardOperateType) {
	go StopPeriodicTask(sourceType, sourceId, operateType)
}

func StopPeriodicTask(sourceType models.SourceType, sourceId string, operateType models.RewardOperateType) error {
	defer func() {
		if err := recover(); err != nil {
			combinedErr := fmt.Errorf("%s\n%s", err, log.Stack(2))
			log.Error("PANIC:%v", combinedErr)
		}
	}()
	task, err := models.GetPeriodicTaskBySourceIdAndType(sourceType, sourceId, operateType)
	if err != nil {
		log.Error("StopPeriodicTask. GetPeriodicTaskBySourceIdAndType error. %v", err)
		return err
	}
	if task == nil {
		log.Info("Periodic task is not exist")
		return nil
	}
	if task.Status == models.PeriodicTaskStatusFinished {
		log.Info("Periodic task is finished")
		return nil
	}
	now := time.Now()
	RunRewardTask(*task, now)
	return models.StopPeriodicTask(task.ID, task.OperateSerialNo, now)
}

func generateOperateSerialNo(operateType models.RewardOperateType, rewardType models.RewardType) (string, error) {
	s, err := GetSerialNoByRedis()
	if err != nil {
		return "", err
	}

	switch operateType {
	case models.OperateTypeIncrease:
		s += "1"
	case models.OperateTypeDecrease:
		s += "2"
	default:
		s += "9"
	}

	switch rewardType {
	case models.RewardTypePoint:
		s += "1"
	default:
		s += "9"
	}

	return s, nil
}