joybuywork/internal/service/provider/s2s/reward.go

package s2s

import (
	"context"
	"joybuywork/internal/app"
	"joybuywork/internal/consts/ckey"
	"joybuywork/internal/repository/cache"
	"joybuywork/internal/repository/dbcheck"
	"joybuywork/internal/repository/model"
	"joybuywork/internal/repository/mtype"
	"joybuywork/internal/repository/withdefault"
	"joybuywork/internal/service/queue/producer"
	"joybuywork/internal/service/queue/types"
	"joybuywork/internal/utils"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"gits.branchcn.com/backend/microcode/plugins/derivative/lock"
)

type Reward struct {
	S2S
	reward *model.Reward

	log *model.RewardPushLog
}

func NewReward(ctx context.Context, reward *model.Reward, opts ...Option) (*Reward, error) {
	r := &Reward{
		reward: reward,
	}

	err := r.init(ctx, reward.ProjectID, reward.URLID, opts...)
	if err != nil {
		return nil, err
	}

	return r, nil
}

func (s *Reward) CreatePush(source string, firstId, secondId, relateId int64, sync int8, param mtype.RewardPushParam) (push *model.RewardPush, err error) {
	// 防止一些并发操作，导致同时推送到客户端
	lc := app.NewRedisLock(s.Ctx, ckey.FissionPushRewardOnce(source, firstId, secondId, relateId), lock.SetExpire(timeOut))
	defer lc.Release()
	if !lc.SoftLock() {
		return
	}

	m := s.GetDao().RewardPush
	push, err = m.WithContext(s.Ctx).Where(
		m.RelateID.Eq(relateId),
		m.FirstID.Eq(firstId),
		m.Source.Eq(source),
		m.SecondID.Eq(secondId),
	).Take()
	if err = dbcheck.DbError(err); err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	if push == nil {
		push = withdefault.RewardPush(&model.RewardPush{
			Source:   source,
			FirstID:  firstId,
			SecondID: secondId,
			RelateID: relateId,
			RewardID: s.reward.ID,
			UserID:   param.UserId,
			Sync:     sync,
		})

		push.SetParam(param)

		err = m.WithContext(s.Ctx).Create(push)
		if err != nil {
			err = s.BE().MysqlError(err)
			return
		}
	} else if push.GetParam() != param {
		push.SetParam(param)
		err = m.WithContext(s.Ctx).Save(push)
		if err != nil {
			err = s.BE().MysqlError(err)
			return
		}
	}

	return
}

func (s *Reward) Send(oldPush *model.RewardPush, sync bool) (push *model.RewardPush, err error) {
	if oldPush.Status == mtype.RewardPushStatusOk {
		return oldPush, nil
	}

	// 防止一些并发操作，导致同时推送到客户端
	lc := app.NewRedisLock(s.Ctx, ckey.FissionPushRewardOnce(oldPush.Source, oldPush.FirstID, oldPush.SecondID, oldPush.RelateID), lock.SetExpire(timeOut))
	defer lc.Release()
	if !lc.SoftLock() {
		return
	}

	push = oldPush
	param := push.GetParam()

	langInfo, err := s._langInfo(param.Lang)
	if err != nil {
		return
	}

	var kv = make(map[string]string, len(s.reward.GetJustKvData()))
	for _, v := range s.reward.GetJustKvData() {
		kv[v.Key] = v.Value
	}

	var campaignHashId string
	if param.CampaignId > 0 {
		cpn, _ := cache.NewCampaign(s.Ctx).FindById(param.CampaignId)
		campaignHashId = cpn.HashID
	}

	var req = mtype.RewardPushLogReq{
		TaskId:         push.ID,
		UserId:         param.UserId,
		CampaignId:     campaignHashId,
		Source:         push.Source,
		RewardId:       param.RelateHashId,
		RewardTitle:    langInfo.Title,
		RewardNum:      param.Num,
		KvData:         kv,
		UserPid:        param.ParentUserId,
		Lang:           param.Lang,
		Title:          param.Title,
		UserPlatform:   utils.SupportPlatform(param.UserPlatform),
		ParentPlatform: utils.SupportPlatform(param.ParentUserPlatform),
		RewardType:     param.RewardType,
	}

	switch push.Source {
	case mtype.RewardPushSourceWall:
		req.WallId = param.SourceHashId
	case mtype.RewardPushSourceCenter:
		req.CenterId = param.SourceHashId
	}

	s.before(push.ID, req)
	body, _ := json.Marshal(req)
	var errMsg string
	resp, code, err := s.request(s.log.ID, body, param.Lang, prefixToken, OldCertToken)
	defer func() {
		s.after(code, resp, errMsg)
	}()

	push.PushNum++

	if strings.HasPrefix(resp, "success") {
		_, nd, _ := strings.Cut(resp, ":")

		push.Status = mtype.RewardPushStatusOk
		push.RespData = nd
		push.FailType = ""
		push.FailMsg = ""

		err = s.GetDao().RewardPush.WithContext(s.Ctx).Save(push)
		if err != nil {
			err = s.BE().MysqlError(err)
			return
		}

		s._notify(push)

		return
	}

	push.Status = mtype.RewardPushStatusFail

	if err != nil {

		val := fmt.Sprintf("%v", err)

		if strings.Contains(val, "missing port in address") {
			push.FailType = mtype.RewardPushFailTypeCli
			push.FailMsg = "回调地址无法正常访问"
		} else {
			push.FailType = mtype.RewardPushFailTypeSys
			push.FailMsg = val
		}

	} else if resp == "" {

		push.FailType = mtype.RewardPushFailTypeCli
		push.FailMsg = "如果处理成功请返回字符串success，当前接收到的信息为空字符串，无法判断当前回调是否正确"

	} else if !strings.HasPrefix(resp, "success") {

		push.FailType = mtype.RewardPushFailTypeCli
		push.FailMsg = fmt.Sprintf("返回无法识别值：%s", resp)

	} else {

		push.FailType = mtype.RewardPushFailTypeUndefined
		push.FailMsg = "未知错误"

	}

	errMsg = push.FailMsg
	err = nil

	if !sync && s._retry(push) {
		m := s.GetDao().RewardPush
		_, err = m.WithContext(s.Ctx).Where(m.ID.Eq(push.ID)).UpdateSimple(m.PushNum.Add(1))
		if err != nil {
			err = s.BE().MysqlError(err)
			return
		}

		return
	}

	err = s.GetDao().RewardPush.WithContext(s.Ctx).Save(push)
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	s._notify(push)

	return
}

func (s *Reward) Async(push *model.RewardPush, delaySec int) {
	task := withdefault.RewardPushTask(&model.RewardPushTask{
		DoTime: time.Now().Add(time.Duration(delaySec) * time.Second).Unix(),
		PushID: push.ID,
	})

	_ = s.GetDao().RewardPushTask.WithContext(s.Ctx).Create(task)
	producer.GlobalRewardPushPub(s.Ctx, delaySec, types.DelayCommonGlobalRewardPush{TaskId: task.ID})
}

func (s *Reward) _retry(push *model.RewardPush) (hasNext bool) {
	//如果没有发送成功并且不是最后一次发送，就创建一个任务
	delaySec, hasDelay := delaySecMap[push.PushNum]
	if !hasDelay {
		return
	}

	s.Async(push, delaySec)

	return true
}

func (s *Reward) _langInfo(lang string) (langInfo *model.RewardLang, err error) {
	m := s.GetDao().RewardLang
	langInfo, err = m.WithContext(s.Ctx).Where(m.RewardID.Eq(s.reward.ID), m.Mark.Eq(lang), m.IsDel.Eq(mtype.No)).Take()
	if err = dbcheck.DbError(err); err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	if langInfo != nil {
		return
	}

	langInfo, err = m.WithContext(s.Ctx).Where(m.RewardID.Eq(s.reward.ID), m.IsDefault.Eq(mtype.Yes), m.IsDel.Eq(mtype.No)).Take()
	if err = dbcheck.DbError(err); err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	return
}

func (s *Reward) before(pushId string, req mtype.RewardPushLogReq) {
	s.log = withdefault.RewardPushLog(&model.RewardPushLog{
		PushID: pushId,
		ReqURL: s.GetUrl(),
	})

	s.log.SetReq(req)

	_ = s.GetDao().RewardPushLog.WithContext(s.Ctx).Create(s.log)
	s.start = time.Now()
}

func (s *Reward) after(code int32, resp, errMsg string) {
	s.log.RespCode = code
	s.log.Resp = resp
	s.log.ErrMsg = errMsg
	s.log.Token = s.GetToken()
	s.log.ReqHeader = s.GetHeader()
	s.log.ReqTime = time.Since(s.start).Milliseconds()

	_ = s.GetDao().RewardPushLog.WithContext(s.Ctx).Save(s.log)
}

func (s *Reward) _notify(push *model.RewardPush) {
	producer.NewDeepworkNotify(s.Ctx).RewardPushStatus(types.DnRewardPushStatus{
		Source:         push.Source,
		SourceId:       push.FirstID,
		SourceSecondId: push.SecondID,
		RelateId:       push.RelateID,
		TaskId:         push.ID,
		Status:         int32(push.Status),
		FailType:       push.FailType,
		FailMsg:        push.FailMsg,
		RewardVal:      push.RespData,
		RewardType:     push.GetParam().RewardType,
	})
}