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/modules/util"
	"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) bool
	Operate(ctx models.RewardOperateContext) error
}

func Send(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)
		}
	}()
	//add lock
	var rewardLock = redis_lock.NewDistributeLock(redis_key.RewardSendLock(ctx.RequestId, ctx.SourceType))
	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, ctx.RequestId)
	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")
	}

	//is limited?
	if operator.IsLimited(ctx) {
		return nil
	}

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

	ctx.SourceId = recordId

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

	//if not a cycle operate,update status to success
	if ctx.CycleIntervalSeconds == 0 {
		updateAwardOperateRecordStatus(ctx.SourceType, ctx.RequestId, models.OperateStatusOperating, models.OperateStatusSucceeded)
	}
	return nil
}

func GetOperator(rewardType string) RewardOperator {
	return RewardOperatorMap[rewardType]
}

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

}

func initAwardOperateRecord(ctx models.RewardOperateContext) (string, error) {
	record := &models.RewardOperateRecord{
		RecordId:             util.UUID(),
		UserId:               ctx.TargetUserId,
		Amount:               ctx.Reward.Amount,
		RewardType:           ctx.Reward.Type,
		SourceType:           ctx.SourceType,
		SourceId:             ctx.SourceId,
		RequestId:            ctx.RequestId,
		OperateType:          ctx.OperateType,
		CycleIntervalSeconds: ctx.CycleIntervalSeconds,
		Status:               models.OperateStatusOperating,
		Remark:               ctx.Remark,
	}
	_, err := models.InsertAwardOperateRecord(record)
	if err != nil {
		return "", err
	}
	return record.RecordId, nil
}

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