package pusher import ( "context" "joybuywork/internal/app" "joybuywork/internal/config" "joybuywork/internal/consts" "joybuywork/internal/consts/ckey" "joybuywork/internal/repository/cache" "joybuywork/internal/repository/dao" "joybuywork/internal/repository/model" "joybuywork/internal/repository/mtype" "joybuywork/internal/service/provider" "joybuywork/internal/service/queue/producer" "joybuywork/internal/service/queue/types" "joybuywork/internal/utils" "joybuywork/internal/utils/blink" "joybuywork/internal/utils/logic" "joybuywork/internal/utils/merr" "fmt" "time" "gits.branchcn.com/backend/microcode/plugins/derivative/lock" ) type UserLine struct { Id int64 DoneAt int64 } type Pusher struct { logic.Context cpn *model.FissionCampaign project *model.Project channelLink map[int64]string // 渠道链接 pg map[string]*pushGroup // 防止重复推送分组 langTitle map[string]string // 语言页面标题 targetNum int64 // 需要邀请的目标数 icon string // 玩法icon } func NewPusher(ctx context.Context, cpn *model.FissionCampaign) (*Pusher, error) { pusher := &Pusher{ cpn: cpn, pg: make(map[string]*pushGroup), langTitle: make(map[string]string), } pusher.Init(ctx, dao.Q, merr.NewBE(ctx)) err := pusher.set() if err != nil { return nil, err } return pusher, nil } // 设置必定需要的参数 func (s *Pusher) set() (err error) { channelM := dao.FissionCampaignChannel channelList, err := channelM.WithContext(s.Ctx).Where( channelM.CampaignID.Eq(s.cpn.ID), channelM.CType.Eq(mtype.FissionCampaignChannelTypeApp), ).Find() if err != nil { err = s.BE().MysqlError(err) return } s.project, err = cache.NewProject(s.Ctx).FindById(s.cpn.ProjectID) if err != nil { err = s.BE().CacheError(err) return } s.channelLink = make(map[int64]string, len(channelList)) for _, v := range channelList { s.channelLink[v.ID] = blink.NewCampaignLink(s.project.Domain, s.cpn.HashID, v.Mark).CampaignShareLink() if v.IsDefault == mtype.Yes && v.IsDel == mtype.No { s.channelLink[0] = s.channelLink[v.ID] } } switch s.cpn.FissionID { case consts.IdCash: taskM := dao.FissionCampaignTask task, err1 := taskM.WithContext(s.Ctx).Where( taskM.CampaignID.Eq(s.cpn.ID), taskM.TaskType.In(mtype.FissionCampaignTaskTypeInstall, mtype.FissionCampaignTaskTypeRegister), taskM.IsDel.Eq(mtype.No), ).Take() if err1 != nil { err = s.BE().MysqlError(err1) return } if task == nil { return } s.targetNum = int64(1 / task.Percent) } return } func (s *Pusher) CompleteTask(user *model.FissionUser) (err error) { if s.beLimit(s.cpn.ProjectID, mtype.FissionCampaignPushMarkCompleteTask, user.SubUserID) { return } inviteNum, err1 := s._inviteNum([]int64{user.ID}) if err1 != nil { return err1 } userMap, err := s._userChannel(s.cpn.ID, []int64{user.ID}) if err != nil { return err } s._push( mtype.FissionCampaignPushMarkCompleteTask, "", "", app.TalkMap.CompeteTask(user.LangMark), userMap[user.ID], 0, inviteNum[user.ID], ) return } // Active 日活推送 func (s *Pusher) Active(t time.Time, markCode, codeVal string, userLines []*UserLine) (err error) { return s.gpPush(mtype.FissionCampaignPushMarkActive, markCode, codeVal, userLines, s._getPg(mtype.FissionCampaignPushMarkActive, t)) } // TaskRest 任务重制推送 func (s *Pusher) TaskRest(t time.Time, markCode, codeVal string, userLines []*UserLine) (err error) { return s.gpPush(mtype.FissionCampaignPushMarkTaskReset, markCode, codeVal, userLines, s._getPg(mtype.FissionCampaignPushMarkTaskReset, t)) } // Offline 离线推送 func (s *Pusher) Offline(t time.Time, markCode, codeVal string, userLines []*UserLine) (err error) { return s.gpPush(mtype.FissionCampaignPushMarkOffline, markCode, codeVal, userLines, s._getPg(mtype.FissionCampaignPushMarkOffline, t)) } // Visit 访问推送 func (s *Pusher) Visit(user *model.FissionUser, t time.Time) (err error) { userMap, err := s._userChannel(s.cpn.ID, []int64{user.ID}) if err != nil { return err } s._push( mtype.FissionCampaignPushMarkVisit, "", "", "", userMap[user.ID], t.Unix(), 0, ) return } // 记录组推送,一些需要防止二次推送的处理的 func (s *Pusher) gpPush(mark, markCode, codeVal string, userLines []*UserLine, pg *pushGroup) (err error) { var userIds []int64 for _, v := range userLines { id := fmt.Sprintf("%d", v.Id) if pg.insideOne(v.DoneAt) && pg.one.Exist(id) { continue } if pg.insideTwo(v.DoneAt) && pg.two.Exist(id) { continue } if pg.insideThree(v.DoneAt) && pg.three.Exist(id) { continue } userIds = append(userIds, v.Id) } inviteNum, err1 := s._inviteNum(userIds) if err1 != nil { return err1 } cpn := s.cpn userMap, err1 := s._userChannel(cpn.ID, userIds) if err1 != nil { return err1 } var oneOk, twoOk, threeOk []string for _, v := range userLines { user, ok := userMap[v.Id] if !ok { continue } langMark := utils.NotExistGetFirst[string](user.LangMark, s.cpn.GetLangMarks()) var desc string switch mark { case mtype.FissionCampaignPushMarkOffline: switch s.cpn.FissionID { case consts.IdKeep: desc = app.TalkMap.DogOffline(langMark) } case mtype.FissionCampaignPushMarkTaskReset: desc = app.TalkMap.TaskReset(langMark) case mtype.FissionCampaignPushMarkActive: desc = app.TalkMap.Active(langMark) } id := fmt.Sprintf("%d", v.Id) if pg.insideThree(v.DoneAt) { threeOk = append(threeOk, id, "1") } else if pg.insideTwo(v.DoneAt) { twoOk = append(twoOk, id, "1") } else if pg.insideOne(v.DoneAt) { oneOk = append(oneOk, id, "1") } s._push(mark, markCode, codeVal, desc, user, v.DoneAt, inviteNum[v.Id]) } if len(threeOk) > 0 { _ = pg.three.Set(threeOk...) } if len(twoOk) > 0 { _ = pg.two.Set(twoOk...) } if len(oneOk) > 0 { _ = pg.one.Set(oneOk...) } return } // 推送到开发平台 func (s *Pusher) _push(mark, markCode, codeVal, desc string, user userChannel, doneAt, inviteNum int64) { producer.NewOplatformOther(s.Ctx).OpoCampaignPush(types.OpoCampaignPush{ ProjectId: s.cpn.ProjectID, FissionId: s.cpn.FissionID, CampaignHashId: s.cpn.HashID, CampaignTitle: s.cpn.Title, CampaignUrl: user.ChannelUrl, PageTitle: s._getLangTitle(user.LangMark), FissionIcon: fmt.Sprintf("%s/%s", config.Config.Upload.Host, consts.GetPushLogo(s.cpn.FissionID, int32(s.cpn.Target))), UserId: user.UserId, Mark: mark, MarkCode: markCode, CodeVal: codeVal, MarkDesc: desc, PushTime: time.Now().Unix(), TaskExpireTime: doneAt, InviteNum: inviteNum, TargetNum: s.targetNum, }, producer.SetHeaderLang(user.LangMark)) } // 获取页面标题 func (s *Pusher) _getLangTitle(userLang string) (langTitle string) { langTitle, ok := s.langTitle[userLang] if ok { return langTitle } lt, err := provider.NewLangTitle(s.Ctx, s.cpn, userLang) if err != nil { return } langTitle, _ = lt.GetPageTitle() s.langTitle[userLang] = langTitle return } // 获取push记录组 func (s *Pusher) _getPg(mark string, t time.Time) *pushGroup { pg, ok := s.pg[mark] if !ok { pg = newPushGroup(s.Ctx, t, s.cpn.ID, mark) s.pg[mark] = pg return pg } return pg } // 邀请人数 func (s *Pusher) _inviteNum(ids []int64) (data map[int64]int64, err error) { data = make(map[int64]int64, len(ids)) m := dao.FissionUserStat list, err := m.WithContext(s.Ctx).Where(m.UserID.In(ids...)).Find() if err != nil { return } for _, v := range list { data[v.UserID] = v.EffectiveNum } return } type userChannel struct { UserId string `json:"user_id"` ChannelUrl string `json:"channel"` LangMark string `json:"lang_mark"` } // 用户渠道 func (s *Pusher) _userChannel(cpnId int64, ids []int64) (data map[int64]userChannel, err error) { m := dao.FissionUser.SubTable(cpnId) list, err := m.WithContext(s.Ctx).Where(m.ID.In(ids...)).Find() if err != nil { return } var channelIds []int64 var channelMap = make(map[int64]string) for _, v := range list { if _, ok := channelMap[v.ChannelID]; !ok { channelMap[v.ChannelID] = "" channelIds = append(channelIds, v.ChannelID) } } data = make(map[int64]userChannel, len(ids)) for _, user := range list { if user.IsView == mtype.Yes { continue } data[user.ID] = userChannel{ UserId: user.SubUserID, ChannelUrl: s.channelLink[user.ChannelID], LangMark: user.LangMark, } } return } func (s *Pusher) beLimit(projectId int64, mark, subUserId string) bool { return !app.NewRedisLock(s.Ctx, ckey.FissionPushLimit(projectId, mark, subUserId), lock.SetExpire(2*time.Hour)).SoftLock() } func (s *Pusher) GetChannelUrl(channelId int64) string { return s.channelLink[channelId] } func (s *Pusher) GetProject() *model.Project { return s.project }