paopao-ce/internal/dao/cache/base.go

// Copyright 2023 ROC. All rights reserved.
// Use of this source code is governed by a MIT style
// license that can be found in the LICENSE file.

package cache

import (
	"bytes"
	"encoding/gob"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/Masterminds/semver/v3"
	"github.com/rocboss/paopao-ce/internal/conf"
	"github.com/rocboss/paopao-ce/internal/core"
	"github.com/rocboss/paopao-ce/internal/core/ms"
	"github.com/rocboss/paopao-ce/pkg/types"
	"github.com/sirupsen/logrus"
)

const (
	_cacheIndexKey = "paopao_index"
)

var (
	_ core.CacheIndexService = (*cacheIndexSrv)(nil)
	_ core.VersionInfo       = (*cacheIndexSrv)(nil)
)

type postsEntry struct {
	key    string
	tweets *ms.IndexTweetList
}

type tweetsCache interface {
	core.VersionInfo
	getTweetsBytes(key string) ([]byte, error)
	setTweetsBytes(key string, bs []byte) error
	delTweets(keys []string) error
	allKeys() ([]string, error)
}

type cacheIndexSrv struct {
	ips core.IndexPostsService
	ams core.AuthorizationManageService

	name               string
	version            *semver.Version
	indexActionCh      chan *core.IndexAction
	cachePostsCh       chan *postsEntry
	cache              tweetsCache
	lastCacheResetTime time.Time
	preventDuration    time.Duration
}

func (s *cacheIndexSrv) IndexPosts(user *ms.User, offset int, limit int) (*ms.IndexTweetList, error) {
	key := s.keyFrom(user, offset, limit)
	posts, err := s.getPosts(key)
	if err == nil {
		logrus.Debugf("cacheIndexSrv.IndexPosts get index posts from cache by key: %s", key)
		return posts, nil
	}

	if posts, err = s.ips.IndexPosts(user, offset, limit); err != nil {
		return nil, err
	}
	logrus.Debugf("cacheIndexSrv.IndexPosts get index posts from database by key: %s", key)
	s.cachePosts(key, posts)
	return posts, nil
}

func (s *cacheIndexSrv) getPosts(key string) (*ms.IndexTweetList, error) {
	data, err := s.cache.getTweetsBytes(key)
	if err != nil {
		logrus.Debugf("cacheIndexSrv.getPosts get posts by key: %s from cache err: %v", key, err)
		return nil, err
	}
	buf := bytes.NewBuffer(data)
	dec := gob.NewDecoder(buf)
	var resp ms.IndexTweetList
	if err := dec.Decode(&resp); err != nil {
		logrus.Debugf("cacheIndexSrv.getPosts get posts from cache in decode err: %v", err)
		return nil, err
	}
	return &resp, nil
}

func (s *cacheIndexSrv) cachePosts(key string, tweets *ms.IndexTweetList) {
	entry := &postsEntry{key: key, tweets: tweets}
	select {
	case s.cachePostsCh <- entry:
		logrus.Debugf("cacheIndexSrv.cachePosts cachePosts by chan of key: %s", key)
	default:
		go func(ch chan<- *postsEntry, entry *postsEntry) {
			logrus.Debugf("cacheIndexSrv.cachePosts cachePosts indexAction by goroutine of key: %s", key)
			ch <- entry
		}(s.cachePostsCh, entry)
	}
}

func (s *cacheIndexSrv) setPosts(entry *postsEntry) {
	var buf bytes.Buffer
	enc := gob.NewEncoder(&buf)
	if err := enc.Encode(entry.tweets); err != nil {
		logrus.Debugf("cacheIndexSrv.setPosts setPosts encode post entry err: %v", err)
		return
	}
	if err := s.cache.setTweetsBytes(entry.key, buf.Bytes()); err != nil {
		logrus.Debugf("cacheIndexSrv.setPosts setPosts set cache err: %v", err)
	}
	logrus.Debugf("cacheIndexSrv.setPosts setPosts set cache by key: %s", entry.key)
}

func (s *cacheIndexSrv) keyFrom(user *ms.User, offset int, limit int) string {
	var userId int64 = -1
	if user != nil {
		userId = user.ID
	}
	return fmt.Sprintf("%s:%d:%d:%d", _cacheIndexKey, userId, offset, limit)
}

func (s *cacheIndexSrv) SendAction(act core.IdxAct, post *ms.Post) {
	action := core.NewIndexAction(act, post)
	select {
	case s.indexActionCh <- action:
		logrus.Debugf("cacheIndexSrv.SendAction send indexAction by chan: %s", act)
	default:
		go func(ch chan<- *core.IndexAction, act *core.IndexAction) {
			logrus.Debugf("cacheIndexSrv.SendAction send indexAction by goroutine: %s", action.Act)
			ch <- act
		}(s.indexActionCh, action)
	}
}

func (s *cacheIndexSrv) startIndexPosts() {
	for {
		select {
		case entry := <-s.cachePostsCh:
			s.setPosts(entry)
		case action := <-s.indexActionCh:
			s.handleIndexAction(action)
		}
	}
}

func (s *cacheIndexSrv) handleIndexAction(action *core.IndexAction) {
	act, post := action.Act, action.Post

	// 创建/删除 私密推文特殊处理
	switch act {
	case core.IdxActCreatePost, core.IdxActDeletePost:
		if post.Visibility == core.PostVisitPrivate {
			s.deleteCacheByUserId(post.UserID, true)
			return
		}
	}

	// 如果在s.preventDuration时间内就清除所有缓存，否则只清除自个儿的缓存
	// TODO: 需要优化只清除受影响的缓存，后续完善
	if time.Since(s.lastCacheResetTime) > s.preventDuration {
		s.deleteCacheByUserId(post.UserID, false)
	} else {
		s.deleteCacheByUserId(post.UserID, true)
	}
}

func (s *cacheIndexSrv) deleteCacheByUserId(id int64, oneself bool) {
	var keys []string
	userId := strconv.FormatInt(id, 10)
	friendSet := ms.FriendSet{}
	if !oneself {
		friendSet = s.ams.MyFriendSet(id)
	}
	friendSet[userId] = types.Empty{}

	// 获取需要删除缓存的key，目前是仅删除自个儿的缓存
	allKeys, err := s.cache.allKeys()
	if err != nil {
		logrus.Debugf("cacheIndexSrv.deleteCacheByUserId userId: %s err:%s", userId, err)
	}
	for _, key := range allKeys {
		keyParts := strings.Split(key, ":")
		if len(keyParts) > 2 && keyParts[0] == _cacheIndexKey {
			if _, ok := friendSet[keyParts[1]]; ok {
				keys = append(keys, key)
			}
		}
	}

	// 执行删缓存
	s.cache.delTweets(keys)
	s.lastCacheResetTime = time.Now()
	logrus.Debugf("cacheIndexSrv.deleteCacheByUserId userId:%s oneself:%t keys:%d", userId, oneself, len(keys))
}

func (s *cacheIndexSrv) Name() string {
	return s.name
}

func (s *cacheIndexSrv) Version() *semver.Version {
	return s.version
}

func newCacheIndexSrv(ips core.IndexPostsService, ams core.AuthorizationManageService, tc tweetsCache) *cacheIndexSrv {
	cacheIndex := &cacheIndexSrv{
		ips:             ips,
		ams:             ams,
		cache:           tc,
		name:            tc.Name(),
		version:         tc.Version(),
		preventDuration: 10 * time.Second,
	}

	// indexActionCh capacity custom configure by conf.yaml need in [10, 10000]
	// or re-compile source to adjust min/max capacity
	capacity := conf.CacheIndexSetting.MaxUpdateQPS
	if capacity < 10 {
		capacity = 10
	} else if capacity > 10000 {
		capacity = 10000
	}
	cacheIndex.indexActionCh = make(chan *core.IndexAction, capacity)
	cacheIndex.cachePostsCh = make(chan *postsEntry, capacity)
	// 启动索引更新器
	go cacheIndex.startIndexPosts()

	return cacheIndex
}