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, }) }