go_dreamfactory/modules/buried/module.go

package buried

import (
	"context"
	"fmt"
	"go_dreamfactory/comm"
	"go_dreamfactory/lego/base"
	"go_dreamfactory/lego/core"
	"go_dreamfactory/lego/sys/event"
	"go_dreamfactory/lego/sys/log"
	"go_dreamfactory/modules"
	"go_dreamfactory/pb"
	cfg "go_dreamfactory/sys/configure/structs"
	"go_dreamfactory/sys/db"
	"time"

	"go.mongodb.org/mongo-driver/bson/primitive"
)

/*
模块名:用户埋点完成条件触发系统
模块描述:用户埋点数据中心管理模块
开发人员:李伟
*/

const moduleName = "埋点统计中心"

type Buried struct {
	modules.ModuleBase
	service     base.IRPCXService
	configure   *configureComp
	modelBuried *modelBuried
}

func NewModule() core.IModule {
	return &Buried{}
}

func (this *Buried) GetType() core.M_Modules {
	return comm.ModuleBuried
}

func (this *Buried) Init(service core.IService, module core.IModule, options core.IModuleOptions) (err error) {
	err = this.ModuleBase.Init(service, module, options)
	this.service = service.(base.IRPCXService)
	return
}

func (this *Buried) Start() (err error) {
	err = this.ModuleBase.Start()
	this.service.RegisterFunctionName(string(comm.Rpc_ModuleBuriedTrigger), this.Rpc_ModuleBuriedTrigger)
	return
}

//装备组件
func (this *Buried) OnInstallComp() {
	this.ModuleBase.OnInstallComp()

	this.configure = this.RegisterComp(new(configureComp)).(*configureComp)
	this.modelBuried = this.RegisterComp(new(modelBuried)).(*modelBuried)
}

// 跨服埋点触发
func (this *Buried) Rpc_ModuleBuriedTrigger(ctx context.Context, args *pb.Rpc_ModuleBuriedTriggerReq, reply *pb.Rpc_ModuleBuriedTriggerResp) (err error) {
	this.Debug("跨服埋点触发!", log.Field{Key: "uid", Value: args.Uid}, log.Field{Key: "burieds", Value: args.Burieds})
	this.trigger(args.Uid, args.Burieds...)
	return
}

//激活数据采集点
func (this *Buried) ActiveCondition(uid string, condiIds ...int32) (err error) {
	var (
		conf    *cfg.GameBuriedCondiData
		bdatas  map[int32]*pb.DBBuried
		chanage []*pb.DBBuried //变化埋点
	)

	if bdatas, err = this.modelBuried.getUserBurieds(uid); err != nil {
		return
	}
	chanage = make([]*pb.DBBuried, 0)
	for _, v := range condiIds {
		if conf, err = this.configure.getburiedcondidata(v); err != nil {
			return
		}
		if bdata, ok := bdatas[conf.Type]; ok {
			if conf.Rtype == rtype2 {
				if item, ok := bdata.Items[v]; !ok {
					bdata.Items[v] = &pb.DBBuriedItem{
						Conid:      v,
						State:      pb.BuriedItemState_Activated,
						Value:      0,
						Statistics: make([]string, 0),
						Timestamp:  time.Now().Unix(),
					}
					chanage = append(chanage, bdata)
				} else {
					item.Value = 0
					item.Statistics = make([]string, 0)
					item.Timestamp = time.Now().Unix()
					item.State = pb.BuriedItemState_Activated
					chanage = append(chanage, bdata)
				}
			}
		}
	}
	if len(chanage) > 0 {
		err = this.modelBuried.updateUserBurieds(uid, chanage)
	}
	return
}

//激活数据采集点
func (this *Buried) CheckCondition(uid string, condiIds ...int32) (condIds []int32, err error) {
	var (
		bdatas map[int32]*pb.DBBuried
		conf   *cfg.GameBuriedCondiData
	)
	if bdatas, err = this.modelBuried.getUserBurieds(uid); err != nil {
		return
	}
	condIds = make([]int32, 0)
	for _, v := range condiIds {
		if conf, err = this.configure.getburiedcondidata(v); err != nil {
			return
		}
		if bdata, ok := bdatas[conf.Type]; ok {
			if data, ok := bdata.Items[v]; ok {
				if data.Value >= conf.Value {
					condIds = append(condIds, v)
				}
			}
		}
	}
	return
}

//校验同时激活
func (this *Buried) CheckAndActiveCondition(uid string, condiIds ...int32) (condIds []int32, err error) {
	var (
		bdatas  map[int32]*pb.DBBuried
		conf    *cfg.GameBuriedCondiData
		chanage []*pb.DBBuried //变化埋点
	)

	if bdatas, err = this.modelBuried.getUserBurieds(uid); err != nil {
		return
	}
	condIds = make([]int32, 0)
	chanage = make([]*pb.DBBuried, 0)
	for _, v := range condiIds {
		if conf, err = this.configure.getburiedcondidata(v); err != nil {
			return
		}
		if bdata, ok := bdatas[conf.Type]; ok {
			if data, ok := bdata.Items[v]; ok {
				if data.Value >= conf.Value {
					condIds = append(condIds, v)
				}
			}
			if conf.Rtype == rtype2 {
				if item, ok := bdata.Items[v]; !ok {
					bdata.Items[v] = &pb.DBBuriedItem{
						Conid:      v,
						State:      pb.BuriedItemState_Activated,
						Value:      0,
						Statistics: make([]string, 0),
						Timestamp:  time.Now().Unix(),
					}
					chanage = append(chanage, bdata)
				} else {
					item.Value = 0
					item.Statistics = make([]string, 0)
					item.Timestamp = time.Now().Unix()
					item.State = pb.BuriedItemState_Activated
					chanage = append(chanage, bdata)
				}
			}
		}
	}
	if len(chanage) > 0 {
		err = this.modelBuried.updateUserBurieds(uid, chanage)
	}
	return
}

//触发埋点
func (this *Buried) TriggerBuried(uid string, burieds ...*pb.BuriedParam) {

	if db.IsCross() {
		var (
			stag string
			err  error
		)
		if stag, err = comm.UidToSTag(uid); err != nil {
			this.Error("远程触发埋点错误!", log.Field{Key: "uid", Value: uid}, log.Field{Key: "err", Value: err.Error()})
			return
		}
		if _, err = this.service.AcrossClusterRpcGo(
			context.Background(),
			stag,
			comm.Service_Worker,
			string(comm.Rpc_ModuleBuriedTrigger),
			pb.Rpc_ModuleBuriedTriggerReq{
				Burieds: burieds,
			},
			nil); err != nil {
			this.Error("远程触发埋点错误!", log.Field{Key: "burieds", Value: burieds}, log.Field{Key: "err", Value: err.Error()})
			return
		}
	} else {
		this.trigger(uid, burieds...)
	}
}

func (this *Buried) trigger(uid string, burieds ...*pb.BuriedParam) {
	var (
		pass           map[*pb.BuriedParam][]*cfg.GameBuriedCondiData = make(map[*pb.BuriedParam][]*cfg.GameBuriedCondiData)
		bconf          *cfg.GameBuriedTypeData
		bdatas         map[int32]*pb.DBBuried
		change         []*pb.DBBuried
		bdata          *pb.DBBuried
		ok             bool
		complete       bool
		completeConIds []int32 //完成id列表
		err            error
	)
	this.Debug("触发埋点!", log.Field{Key: "burieds", Value: burieds})

	lock, _ := this.modelBuried.userlock(uid)
	err = lock.Lock()
	if err != nil {
		this.Error("埋点分布式锁失效 err!", log.Field{Key: "uid", Value: uid}, log.Field{Key: "err", Value: err.Error()})
		return
	}
	defer lock.Unlock()

	for _, buried := range burieds {
		conds := this.configure.getCondiDatas(buried.TaskType)
		if bconf, err = this.configure.getburiedtypedata(buried.TaskType); err != nil {
			this.Error("未找到目标埋点类型配置", log.Field{Key: "type", Value: buried.TaskType})
			continue
		}
		for _, cond := range conds {
			if checkburied(buried, bconf, cond) { //判断此埋点数据是否有效
				if _, ok := pass[buried]; !ok {
					pass[buried] = make([]*cfg.GameBuriedCondiData, 0)
				}
				pass[buried] = append(pass[buried], cond)
			}
		}
	}
	if len(pass) > 0 {
		if bdatas, err = this.modelBuried.getUserBurieds(uid); err != nil {
			return
		}
	}
	completeConIds = make([]int32, 0)
	change = make([]*pb.DBBuried, 0)
	//处理校验通过埋点数据
	for buried, conds := range pass {
		if bconf, err = this.configure.getburiedtypedata(buried.TaskType); err != nil {
			this.Error("未找到目标埋点类型配置", log.Field{Key: "type", Value: buried.TaskType})
			continue
		}
		if bdata, ok = bdatas[int32(buried.TaskType)]; !ok {
			bdatas[int32(buried.TaskType)] = &pb.DBBuried{
				Id:    primitive.NewObjectID().Hex(),
				Uid:   uid,
				Btype: int32(buried.TaskType),
				Items: make(map[int32]*pb.DBBuriedItem),
			}
			bdata = bdatas[int32(buried.TaskType)]
		}
		for _, cond := range conds {
			if cond.Rtype == rtype1 { //创号后入录
				if complete, err = this.updateAndCheckBuried(bconf, bdata, buried, cond, true); complete {
					completeConIds = append(completeConIds, cond.Id)
				}
			} else if cond.Rtype == rtype2 { //任务接取后才会录入 判断用户埋点数据是否存在 不存在等待任务系统调用接口 ActivationBuried 激活
				if complete, err = this.updateAndCheckBuried(bconf, bdata, buried, cond, false); complete {
					completeConIds = append(completeConIds, cond.Id)
				}
			}
		}
		change = append(change, bdatas[bdata.Btype])
	}

	if len(change) > 0 { //同步数据
		if err = this.modelBuried.updateUserBurieds(uid, change); err != nil {
			this.Error("更新用户埋点数据错误!", log.Field{Key: "err", Value: err.Error()})
			return
		}
	}

	//通知事件
	if len(completeConIds) > 0 {
		this.Debug("条件达成通知", log.Field{Key: "ConIds", Value: completeConIds})
		event.TriggerEvent(comm.EventBuriedComplete, uid, completeConIds)
	}
}

//更新并校验完成
func (this *Buried) updateAndCheckBuried(bconf *cfg.GameBuriedTypeData, bdata *pb.DBBuried, collec *pb.BuriedParam, cond *cfg.GameBuriedCondiData, autoActivated bool) (complete bool, err error) {
	var (
		ok    bool
		bitem *pb.DBBuriedItem
	)

	if bitem, ok = bdata.Items[int32(cond.Id)]; !ok {
		if autoActivated { //自动激活
			bdata.Items[int32(cond.Id)] = &pb.DBBuriedItem{
				Conid:      cond.Id,
				State:      pb.BuriedItemState_Activated,
				Value:      0,
				Statistics: make([]string, 0),
				Timestamp:  time.Now().Unix(),
			}
			bitem = bdata.Items[int32(cond.Id)]
		} else {
			return
		}
	}

	if bitem.State == pb.BuriedItemState_Inactivated || bitem.State == pb.BuriedItemState_Freeze { //未激活和冻结 不在处理
		return
	}

	switch bconf.Insert { //数据接入方式
	case overlay: //累加数据
		bitem.Value += collec.Value
	case cover:
		bitem.Value = collec.Value
	case statistics:
		ok = true
		for _, v := range bitem.Statistics {
			if v == collec.Statistics { //已统计过
				ok = false
			}
		}
		if ok {
			bitem.Statistics = append(bitem.Statistics, "")
			bitem.Value = int32(len(bitem.Statistics))
		}
	default:
		err = fmt.Errorf("未知的埋点数据处理类型:%d", bconf.Insert)
		return
	}

	if bitem.Value >= cond.Value { //完成进度
		complete = true
	}
	return
}