openbilibili/app/job/main/relation-cache/service/relation.go

package service

import (
	"context"
	"encoding/json"
	"strings"
	"time"

	"go-common/app/job/main/relation-cache/model"
	relation "go-common/app/service/main/relation/model"
	"go-common/library/log"
	"go-common/library/queue/databus"
	xtime "go-common/library/time"

	"github.com/pkg/errors"
)

const (
	_relationFidTable     = "user_relation_fid_"
	_relationMidTable     = "user_relation_mid_"
	_relationStatTable    = "user_relation_stat_"
	_relationTagUserTable = "user_relation_tag_user_"
)

func (s *Service) relationBinLogproc(ctx context.Context) {
	for msg := range s.relationBinLog.Messages() {
		if err := s.handleRelationBinLog(ctx, msg); err != nil {
			log.Error("Failed to handle relation binlog: %s: %+v", BeautifyMessage(msg), err)
			continue
		}
		log.Info("Succeed to handle relation binlog: %s", BeautifyMessage(msg))
	}
}

func (s *Service) handleRelationBinLog(ctx context.Context, msg *databus.Message) error {
	defer func() {
		if err := msg.Commit(); err != nil {
			log.Error("Failed to commit message: %+v", BeautifyMessage(msg))
			return
		}
	}()

	mu := &model.Message{}
	if err := json.Unmarshal(msg.Value, mu); err != nil {
		return errors.WithStack(err)
	}

	switch {
	case strings.HasPrefix(mu.Table, _relationStatTable):
		if err := s.stat(ctx, mu.Action, mu.New, mu.Old); err != nil {
			return err
		}
	case strings.HasPrefix(mu.Table, _relationMidTable):
		if err := s.relationMid(ctx, mu.Action, mu.New, mu.Old); err != nil {
			return err
		}
	case strings.HasPrefix(mu.Table, _relationFidTable):
		if err := s.relationFid(ctx, mu.Action, mu.New, mu.Old); err != nil {
			return err
		}
	case strings.HasPrefix(mu.Table, _relationTagUserTable):
		if err := s.tagUser(ctx, mu.New); err != nil {
			return err
		}
	}

	return nil
}

// stat
func (s *Service) stat(ctx context.Context, action string, nwMsg []byte, oldMsg []byte) (err error) {
	ms := &model.Stat{}
	if err = json.Unmarshal(nwMsg, ms); err != nil {
		log.Error("json.Unmarshal(%v) error(%v)", nwMsg, err)
		return
	}
	mo := &model.Stat{}
	if len(oldMsg) > 0 {
		if err = json.Unmarshal(oldMsg, mo); err != nil {
			log.Error("json.Unmarshal(%v) error(%v)", oldMsg, err)
			err = nil
		}
	}
	return s.dao.DelStatCache(ctx, ms.Mid)
}

// relationMid
func (s *Service) relationMid(ctx context.Context, action string, nwMsg []byte, oldMsg []byte) error {
	mr := &model.Relation{}
	if err := json.Unmarshal(nwMsg, mr); err != nil {
		return errors.WithStack(err)
	}

	f := &relation.Following{
		Mid:       mr.Fid,
		Attribute: mr.Attribute,
		MTime:     xtime.Time(time.Now().Unix()),
	}
	if err := s.upFollowingCache(ctx, mr.Mid, f); err != nil {
		return err
	}

	return s.dao.DelTagsCache(ctx, mr.Mid)
}

// relationFid
func (s *Service) relationFid(ctx context.Context, action string, nwMsg []byte, oldMsg []byte) error {
	var or *model.Relation
	mr := &model.Relation{}
	if err := json.Unmarshal(nwMsg, mr); err != nil {
		log.Error("json.Unmarshal(%v) error(%v)", nwMsg, err)
		return err
	}

	if len(oldMsg) > 0 {
		or = new(model.Relation)
		if err := json.Unmarshal(oldMsg, or); err != nil {
			log.Error("json.Unmarshal(%v) error(%v)", oldMsg, err)
		}
	}

	return s.dao.DelFollowerCache(ctx, mr.Fid)
}

func (s *Service) tagUser(ctx context.Context, newMsg []byte) (err error) {
	var tags struct {
		Fid int64 `json:"fid"`
		Mid int64 `json:"mid"`
	}
	if err = json.Unmarshal(newMsg, &tags); err != nil {
		log.Error("json.Unmarshal err(%v)", err)
		return
	}
	return s.dao.DelTagsCache(ctx, tags.Mid)
}

func (s *Service) upFollowingCache(c context.Context, mid int64, f *relation.Following) (err error) {
	if f.Attribute == 0 {
		s.dao.DelFollowing(c, mid, f)
	} else {
		if err = s.dao.AddFollowingCache(c, mid, f); err != nil {
			return
		}
	}
	if err = s.dao.DelFollowingCache(c, mid); err != nil {
		return
	}
	return s.dao.DelTagCountCache(c, mid)
}