joybuywork/internal/service/provider/pusher/pusher.go

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
}