feat: add hook system, QR code login, and layered cache

- Implement extensible hook system for content moderation with builtin and AI-powered moderation hooks - Add QR code login feature with SSE for real-time status updates and scan/confirm/cancel workflow - Introduce layered cache with local LRU + Redis backend for improved read performance - Refactor post and comment services to use hook-based moderation instead of direct AI service calls - Add local cache configuration options (size, buckets, ttl)
2026-03-20 12:23:28 +08:00
parent bdfcbdadea
commit 98f0c9f2b6
23 changed files with 2726 additions and 76 deletions
--- a/internal/cache/cache.go
+++ b/internal/cache/cache.go
@@ -4,11 +4,14 @@ import (
 	"cmp"
 	"context"
 	"encoding/json"
+	"errors"
 	"math/rand"
 	"sync"
 	"time"
 )

+var ErrKeyNotFound = errors.New("key not found")
+
 // Cache 缓存接口
 type Cache interface {
 	// Set 设置缓存值，支持TTL
--- a/internal/cache/layered_cache.go
+++ b/internal/cache/layered_cache.go
@@ -0,0 +1,470 @@
+package cache
+
+import (
+	"context"
+	"encoding/json"
+	"strconv"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+
+	redisPkg "carrot_bbs/internal/pkg/redis"
+)
+
+type LayeredCache struct {
+	local     *LRUStriped
+	redis     *RedisCache
+	mu        sync.RWMutex
+	stats     CacheStats
+	enabled   bool
+	keyPrefix string
+}
+
+type CacheStats struct {
+	LocalHits   int64
+	RedisHits   int64
+	Misses      int64
+	Sets        int64
+	Invalidates int64
+}
+
+type LayeredCacheOptions struct {
+	LocalSize       int
+	LocalBuckets    int
+	LocalDefaultTTL time.Duration
+	KeyPrefix       string
+	Enabled         bool
+}
+
+func NewLayeredCache(redisClient *redisPkg.Client, opts *LayeredCacheOptions) *LayeredCache {
+	if opts == nil {
+		opts = &LayeredCacheOptions{
+			LocalSize:       10000,
+			LocalBuckets:    0,
+			LocalDefaultTTL: 5 * time.Minute,
+			KeyPrefix:       "",
+			Enabled:         true,
+		}
+	}
+
+	if opts.LocalSize <= 0 {
+		opts.LocalSize = 10000
+	}
+	if opts.LocalDefaultTTL <= 0 {
+		opts.LocalDefaultTTL = 5 * time.Minute
+	}
+
+	localCache := NewLRUStriped(&LRUStripedOptions{
+		Size:          opts.LocalSize,
+		DefaultExpiry: opts.LocalDefaultTTL,
+		Buckets:       opts.LocalBuckets,
+		Name:          "layered_local",
+	})
+
+	var redisCache *RedisCache
+	if redisClient != nil {
+		redisCache = NewRedisCache(redisClient)
+	}
+
+	return &LayeredCache{
+		local:     localCache,
+		redis:     redisCache,
+		enabled:   opts.Enabled,
+		keyPrefix: opts.KeyPrefix,
+	}
+}
+
+func (c *LayeredCache) Set(key string, value any, ttl time.Duration) {
+	if !c.enabled {
+		return
+	}
+
+	c.mu.Lock()
+	c.stats.Sets++
+	c.mu.Unlock()
+
+	c.local.Set(key, value, ttl)
+
+	if c.redis != nil {
+		c.redis.Set(key, value, ttl)
+	}
+}
+
+func (c *LayeredCache) Get(key string) (any, bool) {
+	if !c.enabled {
+		return nil, false
+	}
+
+	if val, ok := c.local.Get(key); ok {
+		c.mu.Lock()
+		c.stats.LocalHits++
+		c.mu.Unlock()
+		return val, true
+	}
+
+	if c.redis != nil {
+		if val, ok := c.redis.Get(key); ok {
+			c.mu.Lock()
+			c.stats.RedisHits++
+			c.mu.Unlock()
+
+			if ttl := c.getLocalTTL(); ttl > 0 {
+				c.local.Set(key, val, ttl)
+			}
+			return val, true
+		}
+	}
+
+	c.mu.Lock()
+	c.stats.Misses++
+	c.mu.Unlock()
+	return nil, false
+}
+
+func (c *LayeredCache) getLocalTTL() time.Duration {
+	return 30 * time.Second
+}
+
+func (c *LayeredCache) Delete(key string) {
+	if !c.enabled {
+		return
+	}
+
+	c.mu.Lock()
+	c.stats.Invalidates++
+	c.mu.Unlock()
+
+	c.local.Delete(key)
+	if c.redis != nil {
+		c.redis.Delete(key)
+	}
+}
+
+func (c *LayeredCache) DeleteByPrefix(prefix string) {
+	if !c.enabled {
+		return
+	}
+
+	c.mu.Lock()
+	c.stats.Invalidates++
+	c.mu.Unlock()
+
+	c.local.DeleteByPrefix(prefix)
+	if c.redis != nil {
+		c.redis.DeleteByPrefix(prefix)
+	}
+}
+
+func (c *LayeredCache) Clear() {
+	c.local.Clear()
+	if c.redis != nil {
+		c.redis.Clear()
+	}
+	c.mu.Lock()
+	c.stats = CacheStats{}
+	c.mu.Unlock()
+}
+
+func (c *LayeredCache) Exists(key string) bool {
+	if !c.enabled {
+		return false
+	}
+
+	if c.local.Exists(key) {
+		return true
+	}
+
+	if c.redis != nil {
+		return c.redis.Exists(key)
+	}
+
+	return false
+}
+
+func (c *LayeredCache) Increment(key string) int64 {
+	if !c.enabled {
+		return 0
+	}
+
+	if c.redis != nil {
+		val := c.redis.Increment(key)
+		c.local.Delete(key)
+		return val
+	}
+
+	return c.local.Increment(key)
+}
+
+func (c *LayeredCache) IncrementBy(key string, value int64) int64 {
+	if !c.enabled {
+		return 0
+	}
+
+	if c.redis != nil {
+		val := c.redis.IncrementBy(key, value)
+		c.local.Delete(key)
+		return val
+	}
+
+	return c.local.IncrementBy(key, value)
+}
+
+func (c *LayeredCache) GetStats() CacheStats {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.stats
+}
+
+func (c *LayeredCache) GetHitRate() float64 {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	total := c.stats.LocalHits + c.stats.RedisHits + c.stats.Misses
+	if total == 0 {
+		return 0
+	}
+	return float64(c.stats.LocalHits+c.stats.RedisHits) / float64(total)
+}
+
+func (c *LayeredCache) HSet(ctx context.Context, key string, field string, value any) error {
+	if !c.enabled || c.redis == nil {
+		return nil
+	}
+	c.local.Delete(key)
+	return c.redis.HSet(ctx, key, field, value)
+}
+
+func (c *LayeredCache) HMSet(ctx context.Context, key string, values map[string]any) error {
+	if !c.enabled || c.redis == nil {
+		return nil
+	}
+	c.local.Delete(key)
+	return c.redis.HMSet(ctx, key, values)
+}
+
+func (c *LayeredCache) HGet(ctx context.Context, key string, field string) (string, error) {
+	if !c.enabled || c.redis == nil {
+		return "", ErrKeyNotFound
+	}
+	return c.redis.HGet(ctx, key, field)
+}
+
+func (c *LayeredCache) HMGet(ctx context.Context, key string, fields ...string) ([]any, error) {
+	if !c.enabled || c.redis == nil {
+		return nil, ErrKeyNotFound
+	}
+	return c.redis.HMGet(ctx, key, fields...)
+}
+
+func (c *LayeredCache) HGetAll(ctx context.Context, key string) (map[string]string, error) {
+	if !c.enabled || c.redis == nil {
+		return nil, ErrKeyNotFound
+	}
+	return c.redis.HGetAll(ctx, key)
+}
+
+func (c *LayeredCache) HDel(ctx context.Context, key string, fields ...string) error {
+	if !c.enabled || c.redis == nil {
+		return nil
+	}
+	c.local.Delete(key)
+	return c.redis.HDel(ctx, key, fields...)
+}
+
+func (c *LayeredCache) ZAdd(ctx context.Context, key string, score float64, member string) error {
+	if !c.enabled || c.redis == nil {
+		return nil
+	}
+	c.local.Delete(key)
+	return c.redis.ZAdd(ctx, key, score, member)
+}
+
+func (c *LayeredCache) ZRangeByScore(ctx context.Context, key string, min, max string, offset, count int64) ([]string, error) {
+	if !c.enabled || c.redis == nil {
+		return nil, ErrKeyNotFound
+	}
+	return c.redis.ZRangeByScore(ctx, key, min, max, offset, count)
+}
+
+func (c *LayeredCache) ZRevRangeByScore(ctx context.Context, key string, max, min string, offset, count int64) ([]string, error) {
+	if !c.enabled || c.redis == nil {
+		return nil, ErrKeyNotFound
+	}
+	return c.redis.ZRevRangeByScore(ctx, key, max, min, offset, count)
+}
+
+func (c *LayeredCache) ZRem(ctx context.Context, key string, members ...any) error {
+	if !c.enabled || c.redis == nil {
+		return nil
+	}
+	c.local.Delete(key)
+	return c.redis.ZRem(ctx, key, members...)
+}
+
+func (c *LayeredCache) ZCard(ctx context.Context, key string) (int64, error) {
+	if !c.enabled || c.redis == nil {
+		return 0, ErrKeyNotFound
+	}
+	return c.redis.ZCard(ctx, key)
+}
+
+func (c *LayeredCache) Incr(ctx context.Context, key string) (int64, error) {
+	if !c.enabled || c.redis == nil {
+		return 0, ErrKeyNotFound
+	}
+	c.local.Delete(key)
+	return c.redis.Incr(ctx, key)
+}
+
+func (c *LayeredCache) Expire(ctx context.Context, key string, ttl time.Duration) error {
+	if !c.enabled || c.redis == nil {
+		return nil
+	}
+	return c.redis.Expire(ctx, key, ttl)
+}
+
+func (c *LayeredCache) SetLocalOnly(key string, value any, ttl time.Duration) {
+	if !c.enabled {
+		return
+	}
+	c.local.Set(key, value, ttl)
+}
+
+func (c *LayeredCache) GetLocalOnly(key string) (any, bool) {
+	if !c.enabled {
+		return nil, false
+	}
+	return c.local.Get(key)
+}
+
+func (c *LayeredCache) SetRedisOnly(key string, value any, ttl time.Duration) {
+	if !c.enabled || c.redis == nil {
+		return
+	}
+	c.redis.Set(key, value, ttl)
+}
+
+func (c *LayeredCache) GetRedisOnly(key string) (any, bool) {
+	if !c.enabled || c.redis == nil {
+		return nil, false
+	}
+	return c.redis.Get(key)
+}
+
+func (c *LayeredCache) GetRedisClient() *redisPkg.Client {
+	if c.redis == nil {
+		return nil
+	}
+	return c.redis.GetRedisClient()
+}
+
+func GetTypedLayered[T any](c *LayeredCache, key string) (T, bool) {
+	var zero T
+	if !c.enabled {
+		return zero, false
+	}
+
+	raw, ok := c.Get(key)
+	if !ok {
+		return zero, false
+	}
+
+	if typed, ok := raw.(T); ok {
+		return typed, true
+	}
+
+	if str, ok := raw.(string); ok {
+		var out T
+		if err := json.Unmarshal([]byte(str), &out); err != nil {
+			return zero, false
+		}
+		return out, true
+	}
+
+	if data, err := json.Marshal(raw); err == nil {
+		var out T
+		if err := json.Unmarshal(data, &out); err != nil {
+			return zero, false
+		}
+		return out, true
+	}
+
+	return zero, false
+}
+
+func GetOrLoadLayered[T any](
+	c *LayeredCache,
+	key string,
+	ttl time.Duration,
+	loader func() (T, error),
+) (T, error) {
+	if cached, ok := GetTypedLayered[T](c, key); ok {
+		return cached, nil
+	}
+
+	loaded, err := loader()
+	if err != nil {
+		var zero T
+		return zero, err
+	}
+
+	c.Set(key, loaded, ttl)
+	return loaded, nil
+}
+
+func (c *LayeredCache) InvalidateUserCache(userID uint) {
+	c.DeleteByPrefix(cacheKeyPrefixUser + ":")
+	c.Delete(buildUserDetailKey(userID))
+}
+
+func (c *LayeredCache) InvalidatePostCache(postID uint) {
+	c.Delete(buildPostDetailKey(postID))
+	c.DeleteByPrefix(cacheKeyPrefixPostList)
+}
+
+func (c *LayeredCache) InvalidateGroupCache(groupID uint) {
+	c.Delete(buildGroupDetailKey(groupID))
+	c.Delete(buildGroupMembersKey(groupID))
+}
+
+func (c *LayeredCache) InvalidateAllLists() {
+	c.DeleteByPrefix(cacheKeyPrefixPostList)
+	c.DeleteByPrefix(cacheKeyPrefixUserActivity)
+}
+
+var (
+	cacheKeyPrefixUser         = "user"
+	cacheKeyPrefixPostList     = "post_list"
+	cacheKeyPrefixUserActivity = "user_activity"
+)
+
+func buildUserDetailKey(userID uint) string {
+	return "user:detail:" + strconv.FormatUint(uint64(userID), 10)
+}
+
+func buildPostDetailKey(postID uint) string {
+	return "post:detail:" + strconv.FormatUint(uint64(postID), 10)
+}
+
+func buildGroupDetailKey(groupID uint) string {
+	return "group:detail:" + strconv.FormatUint(uint64(groupID), 10)
+}
+
+func buildGroupMembersKey(groupID uint) string {
+	return "group:members:" + strconv.FormatUint(uint64(groupID), 10)
+}
+
+var _ Cache = (*LayeredCache)(nil)
+
+func (c *LayeredCache) logStats() {
+	stats := c.GetStats()
+	zap.L().Debug("LayeredCache stats",
+		zap.Int64("local_hits", stats.LocalHits),
+		zap.Int64("redis_hits", stats.RedisHits),
+		zap.Int64("misses", stats.Misses),
+		zap.Int64("sets", stats.Sets),
+		zap.Int64("invalidates", stats.Invalidates),
+		zap.Float64("hit_rate", c.GetHitRate()),
+	)
+}
--- a/internal/cache/lru.go
+++ b/internal/cache/lru.go
@@ -0,0 +1,205 @@
+package cache
+
+import (
+	"container/list"
+	"encoding/json"
+	"sync"
+	"time"
+)
+
+type LRUOptions struct {
+	Size          int
+	DefaultExpiry time.Duration
+	Name          string
+}
+
+type LRU struct {
+	lock              sync.RWMutex
+	size              int
+	len               int
+	currentGeneration int64
+	evictList         *list.List
+	items             map[string]*list.Element
+	defaultExpiry     time.Duration
+	name              string
+}
+
+type lruEntry struct {
+	key        string
+	value      []byte
+	expires    time.Time
+	generation int64
+}
+
+func NewLRU(opts *LRUOptions) *LRU {
+	if opts.Size <= 0 {
+		opts.Size = 1000
+	}
+	return &LRU{
+		name:          opts.Name,
+		size:          opts.Size,
+		evictList:     list.New(),
+		items:         make(map[string]*list.Element, opts.Size),
+		defaultExpiry: opts.DefaultExpiry,
+	}
+}
+
+func (l *LRU) Set(key string, value any, ttl time.Duration) {
+	if ttl <= 0 {
+		ttl = l.defaultExpiry
+	}
+	l.set(key, value, ttl)
+}
+
+func (l *LRU) set(key string, value any, ttl time.Duration) {
+	var expires time.Time
+	if ttl > 0 {
+		expires = time.Now().Add(ttl)
+	}
+
+	buf, err := json.Marshal(value)
+	if err != nil {
+		return
+	}
+
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	if ent, ok := l.items[key]; ok {
+		l.evictList.MoveToFront(ent)
+		e := ent.Value.(*lruEntry)
+		e.value = buf
+		e.expires = expires
+		if e.generation != l.currentGeneration {
+			e.generation = l.currentGeneration
+			l.len++
+		}
+		return
+	}
+
+	ent := &lruEntry{key, buf, expires, l.currentGeneration}
+	entry := l.evictList.PushFront(ent)
+	l.items[key] = entry
+	l.len++
+
+	if l.evictList.Len() > l.size {
+		l.removeElement(l.evictList.Back())
+	}
+}
+
+func (l *LRU) Get(key string) (any, bool) {
+	val, err := l.getItem(key)
+	if err != nil {
+		return nil, false
+	}
+	return val, true
+}
+
+func (l *LRU) getItem(key string) ([]byte, error) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	ent, ok := l.items[key]
+	if !ok {
+		return nil, ErrKeyNotFound
+	}
+	e := ent.Value.(*lruEntry)
+	if e.generation != l.currentGeneration || (!e.expires.IsZero() && time.Now().After(e.expires)) {
+		l.removeElement(ent)
+		return nil, ErrKeyNotFound
+	}
+	l.evictList.MoveToFront(ent)
+	return e.value, nil
+}
+
+func (l *LRU) Delete(key string) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	if ent, ok := l.items[key]; ok {
+		l.removeElement(ent)
+	}
+}
+
+func (l *LRU) DeleteByPrefix(prefix string) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	for key, ent := range l.items {
+		if len(key) >= len(prefix) && key[:len(prefix)] == prefix {
+			l.removeElement(ent)
+		}
+	}
+}
+
+func (l *LRU) Clear() {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	l.len = 0
+	l.currentGeneration++
+}
+
+func (l *LRU) Exists(key string) bool {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+
+	ent, ok := l.items[key]
+	if !ok {
+		return false
+	}
+	e := ent.Value.(*lruEntry)
+	return e.generation == l.currentGeneration && (e.expires.IsZero() || !time.Now().After(e.expires))
+}
+
+func (l *LRU) Increment(key string) int64 {
+	return l.IncrementBy(key, 1)
+}
+
+func (l *LRU) IncrementBy(key string, value int64) int64 {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	var current int64
+	if ent, ok := l.items[key]; ok {
+		e := ent.Value.(*lruEntry)
+		if e.generation == l.currentGeneration && (e.expires.IsZero() || !time.Now().After(e.expires)) {
+			json.Unmarshal(e.value, &current)
+		}
+	}
+
+	current += value
+	buf, _ := json.Marshal(current)
+
+	if ent, ok := l.items[key]; ok {
+		e := ent.Value.(*lruEntry)
+		e.value = buf
+		l.evictList.MoveToFront(ent)
+	} else {
+		ent := &lruEntry{key, buf, time.Time{}, l.currentGeneration}
+		entry := l.evictList.PushFront(ent)
+		l.items[key] = entry
+		l.len++
+	}
+
+	return current
+}
+
+func (l *LRU) Len() int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return l.len
+}
+
+func (l *LRU) Name() string {
+	return l.name
+}
+
+func (l *LRU) removeElement(e *list.Element) {
+	l.evictList.Remove(e)
+	kv := e.Value.(*lruEntry)
+	if kv.generation == l.currentGeneration {
+		l.len--
+	}
+	delete(l.items, kv.key)
+}
--- a/internal/cache/lru_striped.go
+++ b/internal/cache/lru_striped.go
@@ -0,0 +1,167 @@
+package cache
+
+import (
+	"math"
+	"runtime"
+	"time"
+
+	"github.com/cespare/xxhash/v2"
+)
+
+type LRUStriped struct {
+	buckets []*LRU
+	name    string
+}
+
+type LRUStripedOptions struct {
+	Size          int
+	DefaultExpiry time.Duration
+	Name          string
+	Buckets       int
+}
+
+func NewLRUStriped(opts *LRUStripedOptions) *LRUStriped {
+	if opts.Buckets <= 0 {
+		opts.Buckets = runtime.NumCPU()
+	}
+	if opts.Buckets < 4 {
+		opts.Buckets = 4
+	}
+	if opts.Size < opts.Buckets {
+		opts.Size = opts.Buckets * 100
+	}
+
+	opts.Size += int(math.Ceil(float64(opts.Size) * 10.0 / 100.0))
+	bucketSize := (opts.Size / opts.Buckets) + (opts.Size % opts.Buckets)
+
+	buckets := make([]*LRU, opts.Buckets)
+	for i := 0; i < opts.Buckets; i++ {
+		buckets[i] = NewLRU(&LRUOptions{
+			Size:          bucketSize,
+			DefaultExpiry: opts.DefaultExpiry,
+			Name:          opts.Name,
+		})
+	}
+
+	return &LRUStriped{
+		buckets: buckets,
+		name:    opts.Name,
+	}
+}
+
+func (l *LRUStriped) hashKey(key string) uint64 {
+	return xxhash.Sum64String(key)
+}
+
+func (l *LRUStriped) keyBucket(key string) *LRU {
+	return l.buckets[l.hashKey(key)%uint64(len(l.buckets))]
+}
+
+func (l *LRUStriped) Set(key string, value any, ttl time.Duration) {
+	l.keyBucket(key).Set(key, value, ttl)
+}
+
+func (l *LRUStriped) Get(key string) (any, bool) {
+	return l.keyBucket(key).Get(key)
+}
+
+func (l *LRUStriped) Delete(key string) {
+	l.keyBucket(key).Delete(key)
+}
+
+func (l *LRUStriped) DeleteByPrefix(prefix string) {
+	for _, bucket := range l.buckets {
+		bucket.DeleteByPrefix(prefix)
+	}
+}
+
+func (l *LRUStriped) Clear() {
+	for _, bucket := range l.buckets {
+		bucket.Clear()
+	}
+}
+
+func (l *LRUStriped) Exists(key string) bool {
+	return l.keyBucket(key).Exists(key)
+}
+
+func (l *LRUStriped) Increment(key string) int64 {
+	return l.keyBucket(key).Increment(key)
+}
+
+func (l *LRUStriped) IncrementBy(key string, value int64) int64 {
+	return l.keyBucket(key).IncrementBy(key, value)
+}
+
+func (l *LRUStriped) Len() int {
+	var total int
+	for _, bucket := range l.buckets {
+		total += bucket.Len()
+	}
+	return total
+}
+
+func (l *LRUStriped) Name() string {
+	return l.name
+}
+
+func (l *LRUStriped) DeleteMulti(keys []string) error {
+	var err error
+	for _, key := range keys {
+		l.Delete(key)
+	}
+	return err
+}
+
+func (l *LRUStriped) GetMulti(keys []string, values []any) []error {
+	errs := make([]error, len(values))
+	for i, key := range keys {
+		if val, ok := l.Get(key); ok {
+			values[i] = val
+		} else {
+			errs[i] = ErrKeyNotFound
+		}
+	}
+	return errs
+}
+
+func (l *LRUStriped) SetMulti(items map[string]any, ttl time.Duration) error {
+	for key, value := range items {
+		l.Set(key, value, ttl)
+	}
+	return nil
+}
+
+func (l *LRUStriped) Keys() []string {
+	keys := make([]string, 0)
+	for _, bucket := range l.buckets {
+		bucket.lock.RLock()
+		for key, ent := range bucket.items {
+			e := ent.Value.(*lruEntry)
+			if e.generation == bucket.currentGeneration && (e.expires.IsZero() || !time.Now().After(e.expires)) {
+				keys = append(keys, key)
+			}
+		}
+		bucket.lock.RUnlock()
+	}
+	return keys
+}
+
+type LRUStripedInterface interface {
+	Set(key string, value any, ttl time.Duration)
+	Get(key string) (any, bool)
+	Delete(key string)
+	DeleteByPrefix(prefix string)
+	Clear()
+	Exists(key string) bool
+	Increment(key string) int64
+	IncrementBy(key string, value int64) int64
+	Len() int
+	Name() string
+	DeleteMulti(keys []string) error
+	GetMulti(keys []string, values []any) []error
+	SetMulti(items map[string]any, ttl time.Duration) error
+	Keys() []string
+}
+
+var _ LRUStripedInterface = (*LRUStriped)(nil)