by Robin at 20240308; moment recomm

api/consts/consts.go

@@ -74,6 +74,7 @@ const (
 // redis键前缀
 const (
 	RedisStreamerPrefix = "streamer:" //streamer服务前缀
+	RedisMomentPrefix   = "moment:"   //moment服务前缀
 )
 
 //const PackageRootPath = "C:/Users/PC/Desktop/service"

api/proto/moment/proto/moment_api.go

@@ -140,3 +140,18 @@ type ApiListByIdsResp struct {
 	base.BaseResponse
 	Data *ApiListByIdsData `json:"data"`
 }
+
+// api 推荐
+type ApiRecommListReq struct {
+	base.BaseRequest
+	OpType int64 `json:"op_type"`
+}
+
+type ApiRecommListData struct {
+	RecommList []*ApiMomentVO `json:"recomm_list"`
+}
+
+type ApiRecommListResp struct {
+	base.BaseResponse
+	Data *ApiRecommListData `json:"data"`
+}

app/mix/controller/init.go

@@ -103,6 +103,7 @@ func Init(r *gin.Engine) {
 	apiMomentGroup.POST("list_by_mids", middleware.JSONParamValidator(momentproto.ApiListByMidsReq{}), middleware.JwtAuthenticator(), ApiGetMomentListByMids)
 	apiMomentGroup.POST("thumbs_up", middleware.JSONParamValidator(momentproto.ApiThumbsUpReq{}), middleware.JwtAuthenticator(), ApiThumbsUpMoment)
 	apiMomentGroup.POST("list_by_ids", middleware.JSONParamValidator(momentproto.ApiListByIdsReq{}), middleware.JwtAuthenticator(), ApiGetMomentListByIds)
+	apiMomentGroup.POST("recomm_list", middleware.JSONParamValidator(momentproto.ApiRecommListReq{}), middleware.JwtAuthenticator(), ApiGetMomentRecommList)
 
 	// 足迹
 	// apiFootPrintGroup := r.Group("/api/footprint", PrepareToC())

app/mix/controller/moment_api.go

@@ -182,3 +182,25 @@ func ApiThumbsUpMoment(ctx *gin.Context) {
 
 	ReplyOk(ctx, nil)
 }
+
+func ApiGetMomentRecommList(ctx *gin.Context) {
+	req := ctx.MustGet("client_req").(*momentproto.ApiRecommListReq)
+
+	list, ec := service.DefaultService.ApiGetMomentRecommList(ctx, req)
+	if ec != errcode.ErrCodeMomentSrvOk {
+		logger.Error("OpGetMomentRecommList fail, req: %v, ec: %v", util.ToJson(req), ec)
+		ReplyErrCodeMsg(ctx, ec)
+		return
+	}
+
+	mediaFillableList := make([]mediafiller.MediaFillable, len(list))
+	for i, media := range list {
+		mediaFillableList[i] = media.MediaComp
+	}
+	mediafiller.FillList(ctx, mediaFillableList)
+
+	data := &momentproto.ApiRecommListData{
+		RecommList: list,
+	}
+	ReplyOk(ctx, data)
+}

app/mix/dao/mongo.go

@@ -157,6 +157,9 @@ const (
 
 	DBUserVisitOffset  = "user_visit_offset"
 	COLUserVisitOffset = "user_visit_offset"
+
+	DBUserMomentVisitOffset  = "user_moment_visit_offset"
+	COLUserMomentVisitOffset = "user_moment_visit_offset"
 )
 
 // 商品表
@@ -368,6 +371,11 @@ func (m *Mongo) getColUserVisitOffset() *qmgo.Collection {
 	return m.clientMix.Database(DBUserVisitOffset).Collection(COLUserVisitOffset)
 }
 
+// 用户动态访问偏移量表
+func (m *Mongo) getColUserMomentVisitOffset() *qmgo.Collection {
+	return m.clientMix.Database(DBUserMomentVisitOffset).Collection(COLUserMomentVisitOffset)
+}
+
 // 商品相关
 func (m *Mongo) CreateProduct(ctx *gin.Context, product *dbstruct.Product) error {
 	col := m.getColProduct()
@@ -3344,3 +3352,42 @@ func (m *Mongo) GetUserVisitOffset(ctx *gin.Context, mid int64) (*dbstruct.UserV
 	}
 	return uservisitoffset, err
 }
+
+func (m *Mongo) CreateUserMomentVisitOffset(ctx *gin.Context, usermomentvisitoffset *dbstruct.UserMomentVisitOffset) error {
+	col := m.getColUserMomentVisitOffset()
+	_, err := col.InsertOne(ctx, usermomentvisitoffset)
+	return err
+}
+
+func (m *Mongo) UpdateUserMomentVisitOffset(ctx *gin.Context, usermomentvisitoffset *dbstruct.UserMomentVisitOffset) error {
+	col := m.getColUserMomentVisitOffset()
+	filter := qmgo.M{
+		"_id": usermomentvisitoffset.Id,
+		"ver": usermomentvisitoffset.Ver,
+	}
+	set := qmgo.M{
+		"moment_recomm_offset": usermomentvisitoffset.MomentRecommOffset,
+	}
+	if usermomentvisitoffset.BottomFlag != nil {
+		set["bottom_flag"] = usermomentvisitoffset.GetBottomFlag()
+	}
+	update := qmgo.M{
+		"$set": set,
+	}
+	err := col.UpdateOne(ctx, filter, update)
+	return err
+}
+
+func (m *Mongo) GetUserMomentVisitOffset(ctx *gin.Context, mid int64) (*dbstruct.UserMomentVisitOffset, error) {
+	usermomentvisitoffset := &dbstruct.UserMomentVisitOffset{}
+	col := m.getColUserMomentVisitOffset()
+	query := qmgo.M{
+		"_id": mid,
+	}
+	err := col.Find(ctx, query).One(usermomentvisitoffset)
+	if err == qmgo.ErrNoSuchDocuments {
+		err = nil
+		return nil, err
+	}
+	return usermomentvisitoffset, err
+}

app/mix/service/apiservice.go

@@ -1882,6 +1882,29 @@ func (s *Service) ApiThumbsUpMoment(ctx *gin.Context, req *momentproto.ApiThumbs
 	return
 }
 
+// 推荐
+func (s *Service) ApiGetMomentRecommList(ctx *gin.Context, req *momentproto.ApiRecommListReq) (recommMomentList []*momentproto.ApiMomentVO, ec errcode.ErrCode) {
+
+	ec = errcode.ErrCodeMomentSrvOk
+
+	// 从redis中获取主播列表
+	recommlist, err := s.utilGetMomentRecommList(ctx)
+	if err != nil {
+		logger.Error("utilGetMomentRecommList fail, err: %v", err)
+		ec = errcode.ErrCodeMomentSrvFail
+		return
+	}
+
+	recommMomentList, err = s.utilGetMomentRecommListVO(ctx, recommlist, req.Mid, req.OpType)
+	if err != nil {
+		logger.Error("utilGetUpMomentRecommList fail, err: %v", err)
+		ec = errcode.ErrCodeMomentSrvFail
+		return
+	}
+
+	return
+}
+
 func (s *Service) ApiGetThumbsUpList(ctx *gin.Context, req *thumbsupproto.ApiListReq) (list []*dbstruct.ThumbsUp, ec errcode.ErrCode) {
 	ec = errcode.ErrCodeThumbsUpSrvOk
 

app/mix/service/cronservice.go

@@ -54,6 +54,7 @@ func (s *CronService) Init(c any) (err error) {
 	exec.RegTask("text_audit_batch_his", s.TextAuditBatchHis)
 	//exec.RegTask("clear_content_audit_batch_execution_logs", s.ClearContentAuditBatchExecutionLogs)
 	exec.RegTask("send_contact_customer_services_of_last_minute", s.SendContactCustomerServicesOfLastMinute)
+	exec.RegTask("reload_moment_recomm_list", s.ReloadMomentRecommList)
 	exec.LogHandler(customLogHandle)
 	//注册任务handler
 

app/mix/service/logic/user_moment_visit_offset.go

@@ -0,0 +1,49 @@
+package logic
+
+import (
+	"service/app/mix/dao"
+	"service/dbstruct"
+	"service/library/logger"
+
+	"github.com/gin-gonic/gin"
+)
+
+type UserMomentVisitOffset struct {
+	store *dao.Store
+}
+
+func NewUserMomentVisitOffset(store *dao.Store) (a *UserMomentVisitOffset) {
+	a = &UserMomentVisitOffset{
+		store: store,
+	}
+	return
+}
+
+func (p *UserMomentVisitOffset) OpCreate(ctx *gin.Context, uservisitoffset *dbstruct.UserMomentVisitOffset) error {
+	err := p.store.CreateUserMomentVisitOffset(ctx, uservisitoffset)
+	if err != nil {
+		logger.Error("CreateUserMomentVisitOffset fail, err: %v", err)
+		return err
+	}
+	return nil
+}
+
+func (p *UserMomentVisitOffset) OpUpdate(ctx *gin.Context, uservisitoffset *dbstruct.UserMomentVisitOffset) error {
+	err := p.store.UpdateUserMomentVisitOffset(ctx, uservisitoffset)
+	if err != nil {
+		logger.Error("UpdateUserMomentVisitOffset fail, err: %v", err)
+		return err
+	}
+	return nil
+}
+
+func (p *UserMomentVisitOffset) OpGetUserMomentVisitOffset(ctx *gin.Context, mid int64) (*dbstruct.UserMomentVisitOffset, error) {
+
+	uservisitoffset, err := p.store.GetUserMomentVisitOffset(ctx, mid)
+	if err != nil {
+		logger.Error("GetUserMomentVisitOffset fail, err: %v", err)
+		return nil, err
+	}
+
+	return uservisitoffset, nil
+}

app/mix/service/service.go

@@ -101,6 +101,7 @@ var (
 	_DefaultAppConfig                     *logic.AppConfig
 	_DefaultMomentAuditTask               *logic.MomentAuditTask
 	_DefaultUserVisitOffset               *logic.UserVisitOffset
+	_DefaultUserMomentVisitOffset         *logic.UserMomentVisitOffset
 )
 
 type Service struct {
@@ -178,6 +179,7 @@ func (s *Service) Init(c any) (err error) {
 	_DefaultAppConfig = logic.NewAppConfig(store)
 	_DefaultMomentAuditTask = logic.NewMomentAuditTask(store)
 	_DefaultUserVisitOffset = logic.NewUserVisitOffset(store)
+	_DefaultUserMomentVisitOffset = logic.NewUserMomentVisitOffset(store)
 	return
 }
 

app/mix/service/utilservice.go

@@ -889,3 +889,261 @@ func (s *Service) utilGetStreamerRecommListVO(ctx *gin.Context, recommlist []int
 
 	return
 }
+
+func (s *Service) utilGetMomentRecommList(ctx *gin.Context) (recommlist []int64, err error) {
+
+	// 1.从redis中获取数据
+	err = redis.GetRedisClient().GetObject(consts.RedisMomentPrefix+"recomm_list", &recommlist)
+	if err != nil {
+		logger.Error("Redis read failed : %v", err)
+		return
+	}
+
+	// 2.若redis命中失败，再从数据库查
+	if len(recommlist) == 0 {
+		logger.Error("Redis hit failed, reading recommendation list from mongo...")
+
+		list, err := _DefaultMoment.OpList(ctx, &momentproto.OpListReq{
+			CtLowerBound: goproto.Int64(0),
+		})
+		if err != nil {
+			logger.Error("OpList fail, err: %v", err)
+			return nil, err
+		}
+		recommlist = make([]int64, len(list))
+		for i, moment := range list {
+			recommlist[i] = util.DerefInt64(moment.Id)
+		}
+
+		//若数据库命中成功，则立即加载进redis
+		if len(recommlist) != 0 {
+			err := redis.GetRedisClient().Set(consts.RedisMomentPrefix+"recomm_list", recommlist, 0)
+			if err != nil {
+				logger.Error("Redis cache fail, err: %v", err)
+			}
+		}
+	}
+
+	return
+}
+
+func (s *Service) utilGetInitUserMomentVisitOffset(ctx *gin.Context, mid int64, recommlistLength int64) (err error) {
+	uservisitoffset, err := _DefaultUserMomentVisitOffset.OpGetUserMomentVisitOffset(ctx, mid)
+	if err != nil {
+		logger.Error("OpGetUserMomentVisitOffset fail, err: %v", err)
+		return
+	}
+
+	var execFunc func(*gin.Context, *dbstruct.UserMomentVisitOffset) error
+
+	if uservisitoffset == nil {
+		execFunc = _DefaultUserMomentVisitOffset.OpCreate
+	} else {
+		execFunc = _DefaultUserMomentVisitOffset.OpUpdate
+	}
+
+	// 吞吐量大于等于推荐数组长度，则这次获取后已经触底
+	if consts.MomentRecommThroughput >= recommlistLength {
+		err = execFunc(ctx, &dbstruct.UserMomentVisitOffset{
+			Id:                 mid,
+			MomentRecommOffset: 0,
+			BottomFlag:         goproto.Int64(1),
+			Ver:                0,
+		})
+		if err != nil {
+			logger.Error("_DefaultUserMomentVisitOffset OpCreate fail, err: %v", err)
+			return
+		}
+	} else {
+		err = execFunc(ctx, &dbstruct.UserMomentVisitOffset{
+			Id:                 mid,
+			MomentRecommOffset: consts.MomentRecommThroughput,
+			BottomFlag:         goproto.Int64(0),
+			Ver:                0,
+		})
+		if err != nil {
+			logger.Error("_DefaultUserMomentVisitOffset OpCreate fail, err: %v", err)
+			return
+		}
+	}
+
+	return nil
+}
+
+func (s *Service) utilGetUpUserMomentVisitOffset(ctx *gin.Context, mid int64, recommlistLength int64) (uservisitoffset *dbstruct.UserMomentVisitOffset, err error) {
+	uservisitoffset, err = _DefaultUserMomentVisitOffset.OpGetUserMomentVisitOffset(ctx, mid)
+	if err != nil {
+		logger.Error("OpGetUserMomentVisitOffset fail, err: %v", err)
+		return
+	}
+	if uservisitoffset == nil {
+		// 默认用户刚进推荐页面的第一次操作一定是冷启动操作，游标也由冷启动创建
+		return nil, qmgo.ErrNoSuchDocuments
+	} else {
+		nowoffset := (uservisitoffset.MomentRecommOffset + consts.MomentRecommThroughput) % recommlistLength
+		// 向上操作固定清掉触底标志
+		var bottomFlagPtr *int64 = nil
+		if uservisitoffset.GetBottomFlag() == 1 {
+			bottomFlagPtr = goproto.Int64(0)
+		}
+		err = _DefaultUserMomentVisitOffset.OpUpdate(ctx, &dbstruct.UserMomentVisitOffset{
+			Id:                 uservisitoffset.Id,
+			MomentRecommOffset: nowoffset,
+			BottomFlag:         bottomFlagPtr,
+			Ver:                uservisitoffset.Ver,
+		})
+		if err != nil {
+			logger.Error("_DefaultUserMomentVisitOffset OpUpdate fail, err: %v", err)
+			return
+		}
+	}
+	return
+}
+
+func (s *Service) utilGetDownUserMomentVisitOffset(ctx *gin.Context, mid int64, recommlistLength int64) (uservisitoffset *dbstruct.UserMomentVisitOffset, err error) {
+	uservisitoffset, err = _DefaultUserMomentVisitOffset.OpGetUserMomentVisitOffset(ctx, mid)
+	if err != nil {
+		logger.Error("OpGetUserMomentVisitOffset fail, err: %v", err)
+		return
+	}
+
+	if uservisitoffset == nil {
+		// 默认用户刚进推荐页面的第一次操作一定是冷启动操作，游标也由冷启动创建
+		return nil, qmgo.ErrNoSuchDocuments
+	} else {
+		if uservisitoffset.GetBottomFlag() == 1 {
+			return
+		}
+		offset := uservisitoffset.MomentRecommOffset
+		if offset+consts.MomentRecommThroughput >= recommlistLength {
+			err = _DefaultUserMomentVisitOffset.OpUpdate(ctx, &dbstruct.UserMomentVisitOffset{
+				Id:                 uservisitoffset.Id,
+				MomentRecommOffset: 0,
+				BottomFlag:         goproto.Int64(1),
+				Ver:                uservisitoffset.Ver,
+			})
+			if err != nil {
+				logger.Error("_DefaultUserMomentVisitOffset OpUpdate fail, err: %v", err)
+				return
+			}
+		} else {
+			err = _DefaultUserMomentVisitOffset.OpUpdate(ctx, &dbstruct.UserMomentVisitOffset{
+				Id:                 uservisitoffset.Id,
+				MomentRecommOffset: offset + consts.MomentRecommThroughput,
+				Ver:                uservisitoffset.Ver,
+			})
+			if err != nil {
+				logger.Error("_DefaultUserMomentVisitOffset OpUpdate fail, err: %v", err)
+				return
+			}
+		}
+	}
+	return
+}
+
+// 获取该用户当前在主播推荐数组中的游标和是否已经触底标志
+func (s *Service) utilGetMomentRecommlistOffsetAndBottomeFlag(ctx *gin.Context, recommListLength int64, mid int64, opType int64) (offset int64, bottomFlag int64, err error) {
+	offset = int64(0)
+	var uservisitoffset *dbstruct.UserMomentVisitOffset
+
+	// 初始化操作
+	switch opType {
+	case consts.Recomm_Init:
+		err = s.utilGetInitUserMomentVisitOffset(ctx, mid, int64(recommListLength))
+		if err != nil {
+			logger.Error("utilGetInitUserMomentVisitOffset fail, err: %v", err)
+			return
+		}
+	// 向上滚动操作
+	case consts.Recomm_Up:
+		// 若吞吐量比推荐数组长度小，则正常操作，否则认为游标永远是0
+		if consts.MomentRecommThroughput < recommListLength {
+			uservisitoffset, err = s.utilGetUpUserMomentVisitOffset(ctx, mid, int64(recommListLength))
+			if err != nil {
+				logger.Error("utilGetUpUserMomentVisitOffset fail, err: %v", err)
+				return
+			}
+			offset = uservisitoffset.MomentRecommOffset
+		}
+	// 向下滚动操作
+	case consts.Recomm_Down:
+		// 若吞吐量比推荐数组长度小，则正常操作，否则认为直接触底
+		if consts.MomentRecommThroughput < recommListLength {
+			uservisitoffset, err = s.utilGetDownUserMomentVisitOffset(ctx, mid, int64(recommListLength))
+			if err != nil {
+				logger.Error("utilGetUpUserMomentVisitOffset fail, err: %v", err)
+				return
+			}
+			if uservisitoffset.GetBottomFlag() == 1 {
+				return 0, 1, nil
+			}
+			offset = uservisitoffset.MomentRecommOffset
+		} else {
+			return 0, 1, nil
+		}
+	}
+	return
+}
+
+func (s *Service) utilGetMomentRecommListVO(ctx *gin.Context, recommlist []int64, mid int64, opType int64) (recommMomentList []*momentproto.ApiMomentVO, err error) {
+	// 获取用户游标
+	offset := int64(0)
+	recommListLength := len(recommlist)
+	upperBound := recommListLength
+
+	offset, bottomFlag, err := s.utilGetMomentRecommlistOffsetAndBottomeFlag(ctx, int64(recommListLength), mid, opType)
+	if err != nil {
+		logger.Error("utilGetMomentRecommlistOffsetAndBottomeFlag fail, err: %v", err)
+		return
+	}
+	if bottomFlag == 1 { // 已触底
+		return make([]*momentproto.ApiMomentVO, 0), nil
+	}
+
+	if consts.MomentRecommThroughput < recommListLength {
+		upperBound = consts.MomentRecommThroughput
+	}
+
+	// 根据用户游标查询得到结果
+	midList := make([]int64, 0)
+	for i := 0; i < upperBound; i++ {
+		index := (offset + int64(i)) % int64(recommListLength)
+		midList = append(midList, recommlist[index])
+	}
+
+	list, err := _DefaultMoment.OpListByIds(ctx, &momentproto.OpListByIdsReq{
+		Ids: midList,
+	})
+	if err != nil {
+		logger.Error("_DefaultMoment OpListByIds fail, , err: %v", err)
+		return
+	}
+
+	// 获取访问者的关注列表
+	followMap, err := s.utilGetFollowMap(ctx, mid)
+	if err != nil {
+		logger.Error("utilGetFollowMap fail")
+		return
+	}
+
+	// 填充主播信息
+	vos, ec := s.utilFillMomentsStreamerInfo(ctx, list, consts.InterfaceType_Api)
+	if ec != errcode.ErrCodeMomentSrvOk {
+		logger.Error("utilFillMomentsStreamerInfo fail, , err: %v", err)
+		return
+	}
+	recommMomentList = make([]*momentproto.ApiMomentVO, 0)
+	for _, vo := range vos {
+		apiVO, _ := vo.(*momentproto.ApiMomentVO)
+		// 填充是否关注
+		s.utilFillIsFollowedFillable(ctx, followMap, apiVO)
+		// 填充是否点赞
+		if err = s.utilFillIsThumbedUpFillable(ctx, mid, apiVO); err != nil {
+			logger.Error("utilFillIsThumbedUpFillable fail")
+			return
+		}
+		recommMomentList = append(recommMomentList, apiVO)
+	}
+
+	return
+}

app/mix/service/xxljob_tasks.go

@@ -8,6 +8,7 @@ import (
 	accountproto "service/api/proto/account/proto"
 	contact_customer_service_proto "service/api/proto/contact_customer_service/proto"
 	daily_statementproto "service/api/proto/daily_statement/proto"
+	momentproto "service/api/proto/moment/proto"
 	streamerproto "service/api/proto/streamer/proto"
 	vasproto "service/api/proto/vas/proto"
 	"service/bizcommon/util"
@@ -297,3 +298,26 @@ func (s *CronService) SendContactCustomerServicesOfLastMinute(ctx context.Contex
 
 	return "Message send success"
 }
+
+func (s *CronService) ReloadMomentRecommList(ctx context.Context, param *xxl.RunReq) (msg string) {
+	logger.Info("task %v param: %v log_id: %v", param.ExecutorHandler, param.ExecutorParams, xxl.Int64ToStr(param.LogID))
+	logger.Info("Refreshing recommendation list cached in redis...")
+	list, err := _DefaultMoment.OpList(&gin.Context{}, &momentproto.OpListReq{
+		CtLowerBound: goproto.Int64(0),
+	})
+	if err != nil {
+		logger.Error("OpGetMomentList fail, err: %v", err)
+		return fmt.Sprintf("OpGetMomentList fail, err: %v", err)
+	}
+	momentIdList := make([]int64, len(list))
+	for i, moment := range list {
+		momentIdList[i] = util.DerefInt64(moment.Id)
+	}
+	err = redis.GetRedisClient().Set(consts.RedisMomentPrefix+"recomm_list", momentIdList, 0)
+	if err != nil {
+		logger.Error("Redis cache fail, err: %v", err)
+		return fmt.Sprintf("Redis cache fail, err: %v", err)
+	}
+	logger.Info("Refresh moment recommendation list cached in redis accomplished...")
+	return "Refresh  moment recommendation list cached in redis accomplished"
+}

dbstruct/user_moment_visit_offset.go

@@ -0,0 +1,15 @@
+package dbstruct
+
+type UserMomentVisitOffset struct {
+	Id                 int64  `json:"id" bson:"_id"`                                    // id,用户的mid
+	MomentRecommOffset int64  `json:"moment_recomm_offset" bson:"moment_recomm_offset"` // 动态推荐列表偏移量
+	BottomFlag         *int64 `json:"bottom_flag" bson:"bottom_flag"`                   // 是否触底标志
+	Ver                int64  `json:"ver" bson:"ver"`                                   // 乐观锁
+}
+
+func (p *UserMomentVisitOffset) GetBottomFlag() int64 {
+	if p == nil || p.BottomFlag == nil {
+		return 0
+	}
+	return *p.BottomFlag
+}