From 98f0c9f2b669d16aa6eb8649cf406dfc532f4556 Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Fri, 20 Mar 2026 12:23:28 +0800 Subject: [PATCH] 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) --- cmd/server/wire.go | 2 + cmd/server/wire_gen.go | 11 +- go.mod | 2 +- go.sum | 2 + internal/cache/cache.go | 3 + internal/cache/layered_cache.go | 470 +++++++++++++++++++++++ internal/cache/lru.go | 205 ++++++++++ internal/cache/lru_striped.go | 167 ++++++++ internal/config/config.go | 4 + internal/config/redis.go | 23 +- internal/handler/qrcode_handler.go | 237 ++++++++++++ internal/model/login_log.go | 1 + internal/pkg/hook/hook.go | 405 +++++++++++++++++++ internal/pkg/hook/metrics.go | 122 ++++++ internal/pkg/hook/moderation_hooks.go | 353 +++++++++++++++++ internal/pkg/hook/types.go | 272 +++++++++++++ internal/router/router.go | 21 + internal/service/comment_service.go | 71 ++-- internal/service/post_service.go | 55 +-- internal/service/qrcode_login_service.go | 291 ++++++++++++++ internal/wire/handler.go | 1 + internal/wire/infrastructure.go | 61 ++- internal/wire/service.go | 23 +- 23 files changed, 2726 insertions(+), 76 deletions(-) create mode 100644 internal/cache/layered_cache.go create mode 100644 internal/cache/lru.go create mode 100644 internal/cache/lru_striped.go create mode 100644 internal/handler/qrcode_handler.go create mode 100644 internal/pkg/hook/hook.go create mode 100644 internal/pkg/hook/metrics.go create mode 100644 internal/pkg/hook/moderation_hooks.go create mode 100644 internal/pkg/hook/types.go create mode 100644 internal/service/qrcode_login_service.go diff --git a/cmd/server/wire.go b/cmd/server/wire.go index 2c350fb..6f6f587 100644 --- a/cmd/server/wire.go +++ b/cmd/server/wire.go @@ -45,6 +45,7 @@ func ProvideRouter( adminGroupHandler *handler.AdminGroupHandler, adminDashboardHandler *handler.AdminDashboardHandler, adminLogHandler *handler.AdminLogHandler, + qrcodeHandler *handler.QRCodeHandler, logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, @@ -71,6 +72,7 @@ func ProvideRouter( adminGroupHandler, adminDashboardHandler, adminLogHandler, + qrcodeHandler, logService, activityService, casbinService, diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index fbcf182..a13770f 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -46,9 +46,10 @@ func InitializeApp() (*App, error) { openaiClient := wire.ProvideOpenAIClient(config) postAIService := wire.ProvidePostAIService(openaiClient) transactionManager := wire.ProvideTransactionManager(db) - postService := wire.ProvidePostService(postRepository, systemMessageService, gorseClient, postAIService, cache, transactionManager) + manager := wire.ProvideHookManager() + postService := wire.ProvidePostService(postRepository, systemMessageService, gorseClient, postAIService, cache, transactionManager, manager) commentRepository := repository.NewCommentRepository(db) - commentService := wire.ProvideCommentService(commentRepository, postRepository, systemMessageService, gorseClient, postAIService, cache) + commentService := wire.ProvideCommentService(commentRepository, postRepository, systemMessageService, gorseClient, postAIService, cache, manager) operationLogRepository := repository.NewOperationLogRepository(db) loginLogRepository := repository.NewLoginLogRepository(db) dataChangeLogRepository := repository.NewDataChangeLogRepository(db) @@ -108,7 +109,9 @@ func InitializeApp() (*App, error) { adminDashboardService := wire.ProvideAdminDashboardService(db, userRepository, postRepository, commentRepository, groupRepository, userActivityRepository) adminDashboardHandler := handler.NewAdminDashboardHandler(adminDashboardService) adminLogHandler := handler.NewAdminLogHandler(operationLogService, loginLogService, dataChangeLogService) - router := ProvideRouter(userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, gorseHandler, voteHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, logService, userActivityService, casbinService) + qrCodeLoginService := wire.ProvideQRCodeLoginService(client, hub, jwtService, userService, userActivityService, logService) + qrCodeHandler := handler.NewQRCodeHandler(qrCodeLoginService) + router := ProvideRouter(userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, gorseHandler, voteHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, logService, userActivityService, casbinService) app := NewApp(config, db, router, pushService, server, logger) return app, nil } @@ -138,6 +141,7 @@ func ProvideRouter( adminGroupHandler *handler.AdminGroupHandler, adminDashboardHandler *handler.AdminDashboardHandler, adminLogHandler *handler.AdminLogHandler, + qrcodeHandler *handler.QRCodeHandler, logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, @@ -164,6 +168,7 @@ func ProvideRouter( adminGroupHandler, adminDashboardHandler, adminLogHandler, + qrcodeHandler, logService, activityService, casbinService, diff --git a/go.mod b/go.mod index dd60e5f..272bbc7 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( github.com/alicebob/miniredis/v2 v2.37.0 github.com/casbin/casbin/v3 v3.10.0 github.com/casbin/gorm-adapter/v3 v3.41.0 + github.com/cespare/xxhash/v2 v2.3.0 github.com/disintegration/imaging v1.6.2 github.com/gin-gonic/gin v1.12.0 github.com/golang-jwt/jwt/v5 v5.3.1 @@ -33,7 +34,6 @@ require ( github.com/bytedance/sonic v1.15.0 // indirect github.com/bytedance/sonic/loader v0.5.0 // indirect github.com/casbin/govaluate v1.10.0 // indirect - github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dustin/go-humanize v1.0.1 // indirect diff --git a/go.sum b/go.sum index 66ee4c2..a4e74c7 100644 --- a/go.sum +++ b/go.sum @@ -118,6 +118,8 @@ github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= +github.com/google/subcommands v1.2.0 h1:vWQspBTo2nEqTUFita5/KeEWlUL8kQObDFbub/EN9oE= +github.com/google/subcommands v1.2.0/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3yTrtFlrHVk= github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= diff --git a/internal/cache/cache.go b/internal/cache/cache.go index d5f8b1d..29009a8 100644 --- 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 diff --git a/internal/cache/layered_cache.go b/internal/cache/layered_cache.go new file mode 100644 index 0000000..50a094a --- /dev/null +++ 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()), + ) +} diff --git a/internal/cache/lru.go b/internal/cache/lru.go new file mode 100644 index 0000000..8131eb6 --- /dev/null +++ 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, ¤t) + } + } + + 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) +} diff --git a/internal/cache/lru_striped.go b/internal/cache/lru_striped.go new file mode 100644 index 0000000..c921c85 --- /dev/null +++ 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) diff --git a/internal/config/config.go b/internal/config/config.go index 9d0e739..c3357b0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -72,6 +72,10 @@ func Load(configPath string) (*Config, error) { viper.SetDefault("cache.modules.conversation_ttl", 60) viper.SetDefault("cache.modules.unread_count_ttl", 30) viper.SetDefault("cache.modules.group_members_ttl", 120) + viper.SetDefault("cache.local.enabled", true) + viper.SetDefault("cache.local.size", 10000) + viper.SetDefault("cache.local.buckets", 0) + viper.SetDefault("cache.local.default_ttl", 300) viper.SetDefault("jwt.secret", "your-jwt-secret-key-change-in-production") viper.SetDefault("jwt.access_token_expire", 86400) viper.SetDefault("jwt.refresh_token_expire", 604800) diff --git a/internal/config/redis.go b/internal/config/redis.go index 9ea540a..edbc6b6 100644 --- a/internal/config/redis.go +++ b/internal/config/redis.go @@ -17,13 +17,22 @@ type RedisConfig struct { // CacheConfig 缓存配置 type CacheConfig struct { - Enabled bool `mapstructure:"enabled"` - KeyPrefix string `mapstructure:"key_prefix"` - DefaultTTL int `mapstructure:"default_ttl"` - NullTTL int `mapstructure:"null_ttl"` - JitterRatio float64 `mapstructure:"jitter_ratio"` - DisableFlushDB bool `mapstructure:"disable_flushdb"` - Modules CacheModuleTTL `mapstructure:"modules"` + Enabled bool `mapstructure:"enabled"` + KeyPrefix string `mapstructure:"key_prefix"` + DefaultTTL int `mapstructure:"default_ttl"` + NullTTL int `mapstructure:"null_ttl"` + JitterRatio float64 `mapstructure:"jitter_ratio"` + DisableFlushDB bool `mapstructure:"disable_flushdb"` + Modules CacheModuleTTL `mapstructure:"modules"` + Local LocalCacheConfig `mapstructure:"local"` +} + +// LocalCacheConfig 本地缓存配置 +type LocalCacheConfig struct { + Enabled bool `mapstructure:"enabled"` + Size int `mapstructure:"size"` + Buckets int `mapstructure:"buckets"` + DefaultTTL int `mapstructure:"default_ttl"` } // CacheModuleTTL 缓存模块 TTL 配置 diff --git a/internal/handler/qrcode_handler.go b/internal/handler/qrcode_handler.go new file mode 100644 index 0000000..d0ad844 --- /dev/null +++ b/internal/handler/qrcode_handler.go @@ -0,0 +1,237 @@ +package handler + +import ( + "fmt" + "net/http" + "time" + + "github.com/gin-gonic/gin" + + "carrot_bbs/internal/pkg/response" + "carrot_bbs/internal/pkg/sse" + "carrot_bbs/internal/service" +) + +// QRCodeHandler 二维码登录处理器 +type QRCodeHandler struct { + qrcodeService *service.QRCodeLoginService +} + +// NewQRCodeHandler 创建二维码登录处理器 +func NewQRCodeHandler(qrcodeService *service.QRCodeLoginService) *QRCodeHandler { + return &QRCodeHandler{ + qrcodeService: qrcodeService, + } +} + +// GetQRCode 获取二维码 +// GET /api/v1/auth/qrcode +func (h *QRCodeHandler) GetQRCode(c *gin.Context) { + session, err := h.qrcodeService.CreateSession(c.Request.Context()) + if err != nil { + response.InternalServerError(c, "failed to create qrcode session") + return + } + + // 生成二维码URL (deeplink格式) + qrcodeURL := fmt.Sprintf("carrotbbs://qrcode/login?session_id=%s", session.SessionID) + + response.Success(c, gin.H{ + "session_id": session.SessionID, + "qrcode_url": qrcodeURL, + "expires_in": 300, // 5分钟 + "expires_at": session.ExpiresAt, + }) +} + +// SSEEvents SSE事件流 +// GET /api/v1/auth/qrcode/events?session_id=xxx +func (h *QRCodeHandler) SSEEvents(c *gin.Context) { + sessionID := c.Query("session_id") + if sessionID == "" { + response.BadRequest(c, "session_id is required") + return + } + + ch, cancel, replay := h.qrcodeService.GetSSEHub().Subscribe(sessionID, 0) + defer cancel() + + w := c.Writer + flusher, ok := w.(http.Flusher) + if !ok { + response.InternalServerError(c, "streaming unsupported") + return + } + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + c.Status(http.StatusOK) + flusher.Flush() + + writeEvent := func(ev sse.Event) bool { + data, err := sse.EncodeData(ev) + if err != nil { + return false + } + if _, err := fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", ev.ID, ev.Event, data); err != nil { + return false + } + flusher.Flush() + return true + } + + // 发送历史事件 + for _, ev := range replay { + if !writeEvent(ev) { + return + } + } + + // 心跳 + heartbeat := time.NewTicker(25 * time.Second) + defer heartbeat.Stop() + + for { + select { + case <-c.Request.Context().Done(): + return + case ev, ok := <-ch: + if !ok || !writeEvent(ev) { + return + } + case <-heartbeat.C: + if _, err := fmt.Fprint(w, "event: heartbeat\ndata: {}\n\n"); err != nil { + return + } + flusher.Flush() + } + } +} + +// Scan 扫描二维码 +// POST /api/v1/auth/qrcode/scan +func (h *QRCodeHandler) Scan(c *gin.Context) { + userID := c.GetString("user_id") + if userID == "" { + response.Unauthorized(c, "") + return + } + + type ScanRequest struct { + SessionID string `json:"session_id" binding:"required"` + } + + var req ScanRequest + if err := c.ShouldBindJSON(&req); err != nil { + response.BadRequest(c, err.Error()) + return + } + + if err := h.qrcodeService.Scan(c.Request.Context(), req.SessionID, userID); err != nil { + if err.Error() == "qrcode already scanned" { + response.BadRequest(c, "qrcode already scanned by others") + return + } + if err.Error() == "session not found or expired" { + response.NotFound(c, "qrcode expired") + return + } + response.InternalServerError(c, "failed to scan qrcode") + return + } + + // 获取用户信息返回 + user, err := h.qrcodeService.GetUserService().GetUserByID(c.Request.Context(), userID) + if err != nil { + response.InternalServerError(c, "failed to get user info") + return + } + + response.Success(c, gin.H{ + "user": gin.H{ + "id": user.ID, + "nickname": user.Nickname, + "avatar": user.Avatar, + }, + }) +} + +// Confirm 确认登录 +// POST /api/v1/auth/qrcode/confirm +func (h *QRCodeHandler) Confirm(c *gin.Context) { + userID := c.GetString("user_id") + if userID == "" { + response.Unauthorized(c, "") + return + } + + type ConfirmRequest struct { + SessionID string `json:"session_id" binding:"required"` + } + + var req ConfirmRequest + if err := c.ShouldBindJSON(&req); err != nil { + response.BadRequest(c, err.Error()) + return + } + + ip := c.ClientIP() + userAgent := c.GetHeader("User-Agent") + + result, err := h.qrcodeService.Confirm(c.Request.Context(), req.SessionID, userID, ip, userAgent) + if err != nil { + if err.Error() == "unauthorized" { + response.Unauthorized(c, "not authorized to confirm this login") + return + } + if err.Error() == "session not found or expired" { + response.NotFound(c, "qrcode expired") + return + } + response.InternalServerError(c, "failed to confirm login") + return + } + + response.Success(c, gin.H{ + "success": true, + "user": gin.H{ + "id": result.User.ID, + "username": result.User.Username, + "nickname": result.User.Nickname, + "avatar": result.User.Avatar, + }, + }) +} + +// Cancel 取消登录 +// POST /api/v1/auth/qrcode/cancel +func (h *QRCodeHandler) Cancel(c *gin.Context) { + userID := c.GetString("user_id") + if userID == "" { + response.Unauthorized(c, "") + return + } + + type CancelRequest struct { + SessionID string `json:"session_id" binding:"required"` + } + + var req CancelRequest + if err := c.ShouldBindJSON(&req); err != nil { + response.BadRequest(c, err.Error()) + return + } + + if err := h.qrcodeService.Cancel(c.Request.Context(), req.SessionID, userID); err != nil { + if err.Error() == "unauthorized" { + response.Unauthorized(c, "not authorized to cancel this login") + return + } + response.InternalServerError(c, "failed to cancel login") + return + } + + response.Success(c, gin.H{"success": true}) +} diff --git a/internal/model/login_log.go b/internal/model/login_log.go index 63af3f2..b1af818 100644 --- a/internal/model/login_log.go +++ b/internal/model/login_log.go @@ -47,6 +47,7 @@ const ( LoginTypeSMS LoginType = "sms" LoginTypeEmail LoginType = "email" LoginTypeOAuth LoginType = "oauth" + LoginTypeQRCode LoginType = "qrcode" ) // LoginMethod 登录方式 diff --git a/internal/pkg/hook/hook.go b/internal/pkg/hook/hook.go new file mode 100644 index 0000000..454e063 --- /dev/null +++ b/internal/pkg/hook/hook.go @@ -0,0 +1,405 @@ +package hook + +import ( + "context" + "sync" + "time" + + "go.uber.org/zap" +) + +type HookType string + +const ( + HookPostCreated HookType = "post.created" + HookPostUpdated HookType = "post.updated" + HookPostDeleted HookType = "post.deleted" + HookPostPreModerate HookType = "post.pre_moderate" + HookPostModerated HookType = "post.moderated" + HookCommentCreated HookType = "comment.created" + HookCommentDeleted HookType = "comment.deleted" + HookCommentPreModerate HookType = "comment.pre_moderate" + HookCommentModerated HookType = "comment.moderated" + HookUserCreated HookType = "user.created" + HookUserUpdated HookType = "user.updated" + HookUserDeleted HookType = "user.deleted" + HookUserLogin HookType = "user.login" + HookUserLogout HookType = "user.logout" + HookGroupCreated HookType = "group.created" + HookGroupUpdated HookType = "group.updated" + HookGroupDeleted HookType = "group.deleted" + HookGroupJoined HookType = "group.joined" + HookGroupLeft HookType = "group.left" + HookMessageSent HookType = "message.sent" + HookMessageRead HookType = "message.read" + HookNotificationSent HookType = "notification.sent" + HookVoteCreated HookType = "vote.created" + HookVoteUpdated HookType = "vote.updated" + HookFileUploaded HookType = "file.uploaded" +) + +type Priority int + +const ( + PriorityLowest Priority = 0 + PriorityLow Priority = 25 + PriorityNormal Priority = 50 + PriorityHigh Priority = 75 + PriorityHighest Priority = 100 +) + +type HookContext struct { + Ctx context.Context + HookType HookType + UserID uint + Data any + Metadata map[string]any +} + +type HookFunc func(ctx *HookContext) error + +type Hook struct { + Name string + HookType HookType + Func HookFunc + Priority Priority + Async bool + Timeout time.Duration + Enabled bool +} + +type HookStats struct { + TotalCalls int64 + FailedCalls int64 + TotalTime time.Duration + AverageTime time.Duration + LastCalledAt time.Time +} + +type hookEntry struct { + hook *Hook + stats HookStats + mu sync.Mutex +} + +type Manager struct { + hooks map[HookType][]*hookEntry + mu sync.RWMutex + asyncPool chan struct{} + enabled bool + metrics *Metrics +} + +type ManagerOptions struct { + MaxConcurrentAsync int + Enabled bool +} + +func NewManager(opts *ManagerOptions) *Manager { + if opts == nil { + opts = &ManagerOptions{ + MaxConcurrentAsync: 100, + Enabled: true, + } + } + + if opts.MaxConcurrentAsync <= 0 { + opts.MaxConcurrentAsync = 100 + } + + return &Manager{ + hooks: make(map[HookType][]*hookEntry), + asyncPool: make(chan struct{}, opts.MaxConcurrentAsync), + enabled: opts.Enabled, + metrics: NewMetrics(), + } +} + +func (m *Manager) Register(hook *Hook) { + if hook == nil || hook.Func == nil { + return + } + + if hook.Priority == 0 { + hook.Priority = PriorityNormal + } + if hook.Timeout == 0 { + hook.Timeout = 30 * time.Second + } + hook.Enabled = true + + entry := &hookEntry{ + hook: hook, + stats: HookStats{}, + } + + m.mu.Lock() + defer m.mu.Unlock() + + hooks := m.hooks[hook.HookType] + hooks = append(hooks, entry) + + for i := len(hooks) - 1; i > 0; i-- { + if hooks[i].hook.Priority > hooks[i-1].hook.Priority { + hooks[i], hooks[i-1] = hooks[i-1], hooks[i] + } else { + break + } + } + + m.hooks[hook.HookType] = hooks + zap.L().Debug("Hook registered", + zap.String("name", hook.Name), + zap.String("type", string(hook.HookType)), + zap.Int("priority", int(hook.Priority)), + ) +} + +func (m *Manager) RegisterFunc(hookType HookType, name string, fn HookFunc, opts ...HookOption) { + hook := &Hook{ + Name: name, + HookType: hookType, + Func: fn, + } + for _, opt := range opts { + opt(hook) + } + m.Register(hook) +} + +type HookOption func(*Hook) + +func WithPriority(p Priority) HookOption { + return func(h *Hook) { + h.Priority = p + } +} + +func WithAsync() HookOption { + return func(h *Hook) { + h.Async = true + } +} + +func WithTimeout(timeout time.Duration) HookOption { + return func(h *Hook) { + h.Timeout = timeout + } +} + +func (m *Manager) Trigger(ctx context.Context, hookType HookType, userID uint, data any) error { + return m.TriggerWithMetadata(ctx, hookType, userID, data, nil) +} + +func (m *Manager) TriggerWithMetadata(ctx context.Context, hookType HookType, userID uint, data any, metadata map[string]any) error { + if !m.enabled { + return nil + } + + m.mu.RLock() + entries := m.hooks[hookType] + m.mu.RUnlock() + + if len(entries) == 0 { + return nil + } + + hookCtx := &HookContext{ + Ctx: ctx, + HookType: hookType, + UserID: userID, + Data: data, + Metadata: metadata, + } + + var asyncHooks []*hookEntry + + for _, entry := range entries { + if !entry.hook.Enabled { + continue + } + + if entry.hook.Async { + asyncHooks = append(asyncHooks, entry) + continue + } + + if err := m.executeHook(entry, hookCtx); err != nil { + zap.L().Error("Hook execution failed", + zap.String("name", entry.hook.Name), + zap.String("type", string(hookType)), + zap.Error(err), + ) + } + } + + for _, entry := range asyncHooks { + entry := entry + go func() { + select { + case m.asyncPool <- struct{}{}: + defer func() { <-m.asyncPool }() + if err := m.executeHook(entry, hookCtx); err != nil { + zap.L().Error("Async hook execution failed", + zap.String("name", entry.hook.Name), + zap.String("type", string(hookType)), + zap.Error(err), + ) + } + default: + zap.L().Warn("Async hook pool full, skipping hook", + zap.String("name", entry.hook.Name), + zap.String("type", string(hookType)), + ) + } + }() + } + + return nil +} + +func (m *Manager) executeHook(entry *hookEntry, ctx *HookContext) error { + start := time.Now() + + var err error + if entry.hook.Timeout > 0 { + done := make(chan error, 1) + go func() { + defer func() { + if r := recover(); r != nil { + zap.L().Error("Hook panic recovered", + zap.String("name", entry.hook.Name), + zap.Any("panic", r), + ) + done <- nil + } + }() + done <- entry.hook.Func(ctx) + }() + + select { + case err = <-done: + case <-time.After(entry.hook.Timeout): + err = context.DeadlineExceeded + zap.L().Warn("Hook timeout", + zap.String("name", entry.hook.Name), + zap.Duration("timeout", entry.hook.Timeout), + ) + } + } else { + err = entry.hook.Func(ctx) + } + + duration := time.Since(start) + + entry.mu.Lock() + entry.stats.TotalCalls++ + entry.stats.TotalTime += duration + entry.stats.AverageTime = time.Duration(int64(entry.stats.TotalTime) / entry.stats.TotalCalls) + entry.stats.LastCalledAt = time.Now() + if err != nil { + entry.stats.FailedCalls++ + } + entry.mu.Unlock() + + m.metrics.RecordHookExecution(entry.hook.Name, entry.hook.HookType, duration, err == nil) + + return err +} + +func (m *Manager) Unregister(hookType HookType, name string) { + m.mu.Lock() + defer m.mu.Unlock() + + entries := m.hooks[hookType] + for i, entry := range entries { + if entry.hook.Name == name { + m.hooks[hookType] = append(entries[:i], entries[i+1:]...) + zap.L().Debug("Hook unregistered", + zap.String("name", name), + zap.String("type", string(hookType)), + ) + break + } + } +} + +func (m *Manager) Enable(hookType HookType, name string) { + m.mu.RLock() + entries := m.hooks[hookType] + m.mu.RUnlock() + + for _, entry := range entries { + if entry.hook.Name == name { + entry.hook.Enabled = true + break + } + } +} + +func (m *Manager) Disable(hookType HookType, name string) { + m.mu.RLock() + entries := m.hooks[hookType] + m.mu.RUnlock() + + for _, entry := range entries { + if entry.hook.Name == name { + entry.hook.Enabled = false + break + } + } +} + +func (m *Manager) GetStats(hookType HookType, name string) *HookStats { + m.mu.RLock() + entries := m.hooks[hookType] + m.mu.RUnlock() + + for _, entry := range entries { + if entry.hook.Name == name { + entry.mu.Lock() + stats := entry.stats + entry.mu.Unlock() + return &stats + } + } + return nil +} + +func (m *Manager) GetAllStats() map[HookType]map[string]HookStats { + m.mu.RLock() + defer m.mu.RUnlock() + + result := make(map[HookType]map[string]HookStats) + for hookType, entries := range m.hooks { + result[hookType] = make(map[string]HookStats) + for _, entry := range entries { + entry.mu.Lock() + result[hookType][entry.hook.Name] = entry.stats + entry.mu.Unlock() + } + } + return result +} + +func (m *Manager) GetMetrics() *Metrics { + return m.metrics +} + +func (m *Manager) Clear(hookType HookType) { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.hooks, hookType) +} + +func (m *Manager) ClearAll() { + m.mu.Lock() + defer m.mu.Unlock() + m.hooks = make(map[HookType][]*hookEntry) +} + +func (m *Manager) SetEnabled(enabled bool) { + m.mu.Lock() + defer m.mu.Unlock() + m.enabled = enabled +} diff --git a/internal/pkg/hook/metrics.go b/internal/pkg/hook/metrics.go new file mode 100644 index 0000000..d55de7f --- /dev/null +++ b/internal/pkg/hook/metrics.go @@ -0,0 +1,122 @@ +package hook + +import ( + "sync" + "sync/atomic" + "time" +) + +type Metrics struct { + totalCalls atomic.Int64 + failedCalls atomic.Int64 + totalTime atomic.Int64 + hookMetrics map[string]*hookMetric + mu sync.RWMutex +} + +type hookMetric struct { + calls atomic.Int64 + failures atomic.Int64 + totalTime atomic.Int64 + lastCalled atomic.Int64 +} + +func NewMetrics() *Metrics { + return &Metrics{ + hookMetrics: make(map[string]*hookMetric), + } +} + +func (m *Metrics) RecordHookExecution(name string, hookType HookType, duration time.Duration, success bool) { + m.totalCalls.Add(1) + m.totalTime.Add(int64(duration)) + + if !success { + m.failedCalls.Add(1) + } + + key := string(hookType) + ":" + name + + m.mu.Lock() + metric, ok := m.hookMetrics[key] + if !ok { + metric = &hookMetric{} + m.hookMetrics[key] = metric + } + m.mu.Unlock() + + metric.calls.Add(1) + metric.totalTime.Add(int64(duration)) + metric.lastCalled.Store(time.Now().Unix()) + + if !success { + metric.failures.Add(1) + } +} + +type MetricSnapshot struct { + Name string + Type HookType + Calls int64 + Failures int64 + TotalTime time.Duration + AvgTime time.Duration + LastCalled time.Time +} + +func (m *Metrics) GetTotalCalls() int64 { + return m.totalCalls.Load() +} + +func (m *Metrics) GetFailedCalls() int64 { + return m.failedCalls.Load() +} + +func (m *Metrics) GetTotalTime() time.Duration { + return time.Duration(m.totalTime.Load()) +} + +func (m *Metrics) GetHookMetrics() []MetricSnapshot { + m.mu.RLock() + defer m.mu.RUnlock() + + var snapshots []MetricSnapshot + for key, metric := range m.hookMetrics { + calls := metric.calls.Load() + var avgTime time.Duration + if calls > 0 { + avgTime = time.Duration(metric.totalTime.Load() / calls) + } + + hookType, name := parseKey(key) + snapshots = append(snapshots, MetricSnapshot{ + Name: name, + Type: hookType, + Calls: calls, + Failures: metric.failures.Load(), + TotalTime: time.Duration(metric.totalTime.Load()), + AvgTime: avgTime, + LastCalled: time.Unix(metric.lastCalled.Load(), 0), + }) + } + return snapshots +} + +func parseKey(key string) (HookType, string) { + for i := 0; i < len(key); i++ { + if key[i] == ':' { + return HookType(key[:i]), key[i+1:] + } + } + return HookType(key), "" +} + +func (m *Metrics) Reset() { + m.totalCalls.Store(0) + m.failedCalls.Store(0) + m.totalTime.Store(0) + + m.mu.Lock() + m.hookMetrics = make(map[string]*hookMetric) + m.mu.Unlock() +} diff --git a/internal/pkg/hook/moderation_hooks.go b/internal/pkg/hook/moderation_hooks.go new file mode 100644 index 0000000..06e24a6 --- /dev/null +++ b/internal/pkg/hook/moderation_hooks.go @@ -0,0 +1,353 @@ +package hook + +import ( + "context" + "strings" + "time" + + "carrot_bbs/internal/repository" + + "go.uber.org/zap" +) + +type PostModerationHook struct { + name string + postAIService PostAIService + postRepo *repository.PostRepository + strictMode bool +} + +type PostAIService interface { + IsEnabled() bool + ModeratePost(ctx context.Context, title, content string, images []string) error + ModerateComment(ctx context.Context, content string, images []string) error +} + +func NewPostModerationHook(postAIService PostAIService, postRepo *repository.PostRepository, strictMode bool) *PostModerationHook { + return &PostModerationHook{ + name: "moderation.post.ai", + postAIService: postAIService, + postRepo: postRepo, + strictMode: strictMode, + } +} + +func (h *PostModerationHook) Register(manager *Manager) { + manager.Register(&Hook{ + Name: h.name, + HookType: HookPostPreModerate, + Priority: PriorityHighest, + Func: h.Execute, + Async: false, // 必须同步执行，因为调用方需要审核结果 + Timeout: 30 * time.Second, + }) +} + +func (h *PostModerationHook) Execute(ctx *HookContext) error { + data, ok := ctx.Data.(*PostModerateHookData) + if !ok { + return nil + } + + result, ok := ctx.Metadata["result"].(*ModerationResult) + if !ok || result == nil { + return nil + } + + if h.postAIService == nil || !h.postAIService.IsEnabled() { + result.Approved = true + result.ReviewedBy = "system" + zap.L().Debug("AI moderation disabled, auto approve post", + zap.String("post_id", data.PostID), + ) + return nil + } + + err := h.postAIService.ModeratePost(ctx.Ctx, data.Title, data.Content, data.Images) + if err != nil { + if rejectErr, ok := err.(interface{ UserMessage() string }); ok { + result.Approved = false + result.RejectReason = rejectErr.UserMessage() + result.ReviewedBy = "ai" + zap.L().Info("Post rejected by AI moderation", + zap.String("post_id", data.PostID), + zap.String("reason", result.RejectReason), + ) + return nil + } + + if h.strictMode { + result.Approved = false + result.Error = err + result.ReviewedBy = "ai" + return nil + } + + result.Approved = true + result.ReviewedBy = "system" + zap.L().Warn("AI moderation error, fallback approve", + zap.String("post_id", data.PostID), + zap.Error(err), + ) + return nil + } + + result.Approved = true + result.ReviewedBy = "ai" + zap.L().Debug("Post approved by AI moderation", + zap.String("post_id", data.PostID), + ) + return nil +} + +type CommentModerationHook struct { + name string + postAIService PostAIService + commentRepo *repository.CommentRepository + strictMode bool +} + +func NewCommentModerationHook(postAIService PostAIService, commentRepo *repository.CommentRepository, strictMode bool) *CommentModerationHook { + return &CommentModerationHook{ + name: "moderation.comment.ai", + postAIService: postAIService, + commentRepo: commentRepo, + strictMode: strictMode, + } +} + +func (h *CommentModerationHook) Register(manager *Manager) { + manager.Register(&Hook{ + Name: h.name, + HookType: HookCommentPreModerate, + Priority: PriorityHighest, + Func: h.Execute, + Async: false, // 必须同步执行，因为调用方需要审核结果 + Timeout: 30 * time.Second, + }) +} + +func (h *CommentModerationHook) Execute(ctx *HookContext) error { + data, ok := ctx.Data.(*CommentModerateHookData) + if !ok { + return nil + } + + result, ok := ctx.Metadata["result"].(*ModerationResult) + if !ok || result == nil { + return nil + } + + if h.postAIService == nil || !h.postAIService.IsEnabled() { + result.Approved = true + result.ReviewedBy = "system" + zap.L().Debug("AI moderation disabled, auto approve comment", + zap.String("comment_id", data.CommentID), + ) + return nil + } + + err := h.postAIService.ModerateComment(ctx.Ctx, data.Content, data.Images) + if err != nil { + if rejectErr, ok := err.(interface{ UserMessage() string }); ok { + result.Approved = false + result.RejectReason = rejectErr.UserMessage() + result.ReviewedBy = "ai" + zap.L().Info("Comment rejected by AI moderation", + zap.String("comment_id", data.CommentID), + zap.String("reason", result.RejectReason), + ) + return nil + } + + if h.strictMode { + result.Approved = false + result.Error = err + result.ReviewedBy = "ai" + return nil + } + + result.Approved = true + result.ReviewedBy = "system" + zap.L().Warn("AI comment moderation error, fallback approve", + zap.String("comment_id", data.CommentID), + zap.Error(err), + ) + return nil + } + + result.Approved = true + result.ReviewedBy = "ai" + zap.L().Debug("Comment approved by AI moderation", + zap.String("comment_id", data.CommentID), + ) + return nil +} + +type SensitiveWordHook struct { + name string + sensitive SensitiveService + replaceStr string +} + +type SensitiveService interface { + Check(ctx context.Context, text string) (bool, []string) + Replace(ctx context.Context, text string, repl string) string +} + +func NewSensitiveWordHook(sensitive SensitiveService, replaceStr string) *SensitiveWordHook { + return &SensitiveWordHook{ + name: "moderation.sensitive_word", + sensitive: sensitive, + replaceStr: replaceStr, + } +} + +func (h *SensitiveWordHook) RegisterPostHook(manager *Manager) { + manager.Register(&Hook{ + Name: h.name + ".post", + HookType: HookPostPreModerate, + Priority: PriorityHigh, + Func: h.executePost, + Async: false, + }) +} + +func (h *SensitiveWordHook) RegisterCommentHook(manager *Manager) { + manager.Register(&Hook{ + Name: h.name + ".comment", + HookType: HookCommentPreModerate, + Priority: PriorityHigh, + Func: h.executeComment, + Async: false, + }) +} + +func (h *SensitiveWordHook) executePost(ctx *HookContext) error { + if h.sensitive == nil { + return nil + } + + data, ok := ctx.Data.(*PostModerateHookData) + if !ok { + return nil + } + + result, ok := ctx.Metadata["result"].(*ModerationResult) + if !ok || result == nil { + return nil + } + + if result.RejectReason != "" { + return nil + } + + hasSensitive, words := h.sensitive.Check(ctx.Ctx, data.Title+" "+data.Content) + if hasSensitive { + result.Approved = false + result.RejectReason = "内容包含敏感词: " + strings.Join(words, ", ") + result.ReviewedBy = "sensitive_word" + zap.L().Info("Post rejected by sensitive word filter", + zap.String("post_id", data.PostID), + zap.Strings("words", words), + ) + } + + return nil +} + +func (h *SensitiveWordHook) executeComment(ctx *HookContext) error { + if h.sensitive == nil { + return nil + } + + data, ok := ctx.Data.(*CommentModerateHookData) + if !ok { + return nil + } + + result, ok := ctx.Metadata["result"].(*ModerationResult) + if !ok || result == nil { + return nil + } + + if result.RejectReason != "" { + return nil + } + + hasSensitive, words := h.sensitive.Check(ctx.Ctx, data.Content) + if hasSensitive { + result.Approved = false + result.RejectReason = "内容包含敏感词: " + strings.Join(words, ", ") + result.ReviewedBy = "sensitive_word" + zap.L().Info("Comment rejected by sensitive word filter", + zap.String("comment_id", data.CommentID), + zap.Strings("words", words), + ) + } + + return nil +} + +type ModerationHooks struct { + postAIService PostAIService + sensitiveService SensitiveService + postRepo *repository.PostRepository + commentRepo *repository.CommentRepository + strictMode bool + replaceStr string +} + +func NewModerationHooks( + postAIService PostAIService, + sensitiveService SensitiveService, + postRepo *repository.PostRepository, + commentRepo *repository.CommentRepository, + strictMode bool, + replaceStr string, +) *ModerationHooks { + return &ModerationHooks{ + postAIService: postAIService, + sensitiveService: sensitiveService, + postRepo: postRepo, + commentRepo: commentRepo, + strictMode: strictMode, + replaceStr: replaceStr, + } +} + +func (m *ModerationHooks) RegisterAll(manager *Manager) { + postHook := NewPostModerationHook(m.postAIService, m.postRepo, m.strictMode) + postHook.Register(manager) + + commentHook := NewCommentModerationHook(m.postAIService, m.commentRepo, m.strictMode) + commentHook.Register(manager) + + if m.sensitiveService != nil { + sensitiveHook := NewSensitiveWordHook(m.sensitiveService, m.replaceStr) + sensitiveHook.RegisterPostHook(manager) + sensitiveHook.RegisterCommentHook(manager) + } + + zap.L().Info("Registered moderation hooks", + zap.Bool("ai_enabled", m.postAIService != nil && m.postAIService.IsEnabled()), + zap.Bool("sensitive_enabled", m.sensitiveService != nil), + zap.Bool("strict_mode", m.strictMode), + ) +} + +func (m *ModerationHooks) ModeratePost(ctx context.Context, manager *Manager, data *PostModerateHookData) *ModerationResult { + result := &ModerationResult{} + manager.TriggerWithMetadata(ctx, HookPostPreModerate, data.AuthorID, data, map[string]any{ + "result": result, + }) + return result +} + +func (m *ModerationHooks) ModerateComment(ctx context.Context, manager *Manager, data *CommentModerateHookData) *ModerationResult { + result := &ModerationResult{} + manager.TriggerWithMetadata(ctx, HookCommentPreModerate, data.AuthorID, data, map[string]any{ + "result": result, + }) + return result +} diff --git a/internal/pkg/hook/types.go b/internal/pkg/hook/types.go new file mode 100644 index 0000000..1db7522 --- /dev/null +++ b/internal/pkg/hook/types.go @@ -0,0 +1,272 @@ +package hook + +import ( + "context" + + "go.uber.org/zap" +) + +type PostHookData struct { + PostID uint + Title string + Content string + AuthorID uint + GroupID *uint + IsPublished bool +} + +type CommentHookData struct { + CommentID uint + PostID uint + Content string + AuthorID uint + ParentID *uint +} + +type UserHookData struct { + UserID uint + Username string + Email string + Nickname string +} + +type GroupHookData struct { + GroupID uint + Name string + CreatorID uint +} + +type MessageHookData struct { + MessageID uint + SenderID uint + ReceiverID uint + Content string +} + +type NotificationHookData struct { + NotificationID uint + UserID uint + Type string + Title string + Content string +} + +type VoteHookData struct { + VoteID uint + PostID uint + VoterID uint + OptionID uint +} + +type FileHookData struct { + FileID uint + FileName string + FileSize int64 + Uploader uint + MimeType string +} + +type PostModerateHookData struct { + PostID string + Title string + Content string + Images []string + AuthorID uint + GroupID *uint +} + +type PostModeratedHookData struct { + PostID string + AuthorID uint + Approved bool + RejectReason string + ReviewedBy string +} + +type CommentModerateHookData struct { + CommentID string + PostID string + Content string + Images []string + AuthorID uint + ParentID *string +} + +type CommentModeratedHookData struct { + CommentID string + PostID string + AuthorID uint + Approved bool + RejectReason string + ReviewedBy string +} + +type ModerationResult struct { + Approved bool + RejectReason string + ReviewedBy string + Error error +} + +type BuiltinHooks struct { + manager *Manager +} + +func NewBuiltinHooks(manager *Manager) *BuiltinHooks { + return &BuiltinHooks{manager: manager} +} + +func (b *BuiltinHooks) RegisterAll() { + b.registerLoggingHooks() + b.registerCacheInvalidationHooks() +} + +func (b *BuiltinHooks) registerLoggingHooks() { + b.manager.RegisterFunc(HookPostCreated, "log.post.created", func(ctx *HookContext) error { + if data, ok := ctx.Data.(*PostHookData); ok { + zap.L().Info("Post created", + zap.Uint("post_id", data.PostID), + zap.Uint("author_id", data.AuthorID), + zap.String("title", data.Title), + zap.Uint("user_id", ctx.UserID), + ) + } + return nil + }, WithAsync(), WithPriority(PriorityLowest)) + + b.manager.RegisterFunc(HookUserLogin, "log.user.login", func(ctx *HookContext) error { + if data, ok := ctx.Data.(*UserHookData); ok { + zap.L().Info("User logged in", + zap.Uint("user_id", data.UserID), + zap.String("username", data.Username), + ) + } + return nil + }, WithAsync()) + + b.manager.RegisterFunc(HookCommentCreated, "log.comment.created", func(ctx *HookContext) error { + if data, ok := ctx.Data.(*CommentHookData); ok { + zap.L().Info("Comment created", + zap.Uint("comment_id", data.CommentID), + zap.Uint("post_id", data.PostID), + zap.Uint("author_id", data.AuthorID), + ) + } + return nil + }, WithAsync(), WithPriority(PriorityLowest)) +} + +func (b *BuiltinHooks) registerCacheInvalidationHooks() { +} + +type HookTriggerHelper struct { + manager *Manager +} + +func NewHookTriggerHelper(manager *Manager) *HookTriggerHelper { + return &HookTriggerHelper{manager: manager} +} + +func (h *HookTriggerHelper) TriggerPostCreated(ctx context.Context, userID uint, data *PostHookData) { + h.manager.Trigger(ctx, HookPostCreated, userID, data) +} + +func (h *HookTriggerHelper) TriggerPostUpdated(ctx context.Context, userID uint, data *PostHookData) { + h.manager.Trigger(ctx, HookPostUpdated, userID, data) +} + +func (h *HookTriggerHelper) TriggerPostDeleted(ctx context.Context, userID uint, postID uint) { + h.manager.Trigger(ctx, HookPostDeleted, userID, postID) +} + +func (h *HookTriggerHelper) TriggerCommentCreated(ctx context.Context, userID uint, data *CommentHookData) { + h.manager.Trigger(ctx, HookCommentCreated, userID, data) +} + +func (h *HookTriggerHelper) TriggerCommentDeleted(ctx context.Context, userID uint, commentID uint) { + h.manager.Trigger(ctx, HookCommentDeleted, userID, commentID) +} + +func (h *HookTriggerHelper) TriggerUserCreated(ctx context.Context, data *UserHookData) { + h.manager.Trigger(ctx, HookUserCreated, data.UserID, data) +} + +func (h *HookTriggerHelper) TriggerUserUpdated(ctx context.Context, userID uint, data *UserHookData) { + h.manager.Trigger(ctx, HookUserUpdated, userID, data) +} + +func (h *HookTriggerHelper) TriggerUserDeleted(ctx context.Context, userID uint) { + h.manager.Trigger(ctx, HookUserDeleted, userID, userID) +} + +func (h *HookTriggerHelper) TriggerUserLogin(ctx context.Context, userID uint, data *UserHookData) { + h.manager.Trigger(ctx, HookUserLogin, userID, data) +} + +func (h *HookTriggerHelper) TriggerUserLogout(ctx context.Context, userID uint) { + h.manager.Trigger(ctx, HookUserLogout, userID, userID) +} + +func (h *HookTriggerHelper) TriggerGroupCreated(ctx context.Context, userID uint, data *GroupHookData) { + h.manager.Trigger(ctx, HookGroupCreated, userID, data) +} + +func (h *HookTriggerHelper) TriggerGroupUpdated(ctx context.Context, userID uint, data *GroupHookData) { + h.manager.Trigger(ctx, HookGroupUpdated, userID, data) +} + +func (h *HookTriggerHelper) TriggerGroupDeleted(ctx context.Context, userID uint, groupID uint) { + h.manager.Trigger(ctx, HookGroupDeleted, userID, groupID) +} + +func (h *HookTriggerHelper) TriggerGroupJoined(ctx context.Context, userID uint, groupID uint) { + h.manager.Trigger(ctx, HookGroupJoined, userID, groupID) +} + +func (h *HookTriggerHelper) TriggerGroupLeft(ctx context.Context, userID uint, groupID uint) { + h.manager.Trigger(ctx, HookGroupLeft, userID, groupID) +} + +func (h *HookTriggerHelper) TriggerMessageSent(ctx context.Context, userID uint, data *MessageHookData) { + h.manager.Trigger(ctx, HookMessageSent, userID, data) +} + +func (h *HookTriggerHelper) TriggerMessageRead(ctx context.Context, userID uint, messageID uint) { + h.manager.Trigger(ctx, HookMessageRead, userID, messageID) +} + +func (h *HookTriggerHelper) TriggerNotificationSent(ctx context.Context, userID uint, data *NotificationHookData) { + h.manager.Trigger(ctx, HookNotificationSent, userID, data) +} + +func (h *HookTriggerHelper) TriggerVoteCreated(ctx context.Context, userID uint, data *VoteHookData) { + h.manager.Trigger(ctx, HookVoteCreated, userID, data) +} + +func (h *HookTriggerHelper) TriggerVoteUpdated(ctx context.Context, userID uint, data *VoteHookData) { + h.manager.Trigger(ctx, HookVoteUpdated, userID, data) +} + +func (h *HookTriggerHelper) TriggerFileUploaded(ctx context.Context, userID uint, data *FileHookData) { + h.manager.Trigger(ctx, HookFileUploaded, userID, data) +} + +func (h *HookTriggerHelper) TriggerPostPreModerate(ctx context.Context, userID uint, data *PostModerateHookData, result *ModerationResult) { + h.manager.TriggerWithMetadata(ctx, HookPostPreModerate, userID, data, map[string]any{ + "result": result, + }) +} + +func (h *HookTriggerHelper) TriggerPostModerated(ctx context.Context, userID uint, data *PostModeratedHookData) { + h.manager.Trigger(ctx, HookPostModerated, userID, data) +} + +func (h *HookTriggerHelper) TriggerCommentPreModerate(ctx context.Context, userID uint, data *CommentModerateHookData, result *ModerationResult) { + h.manager.TriggerWithMetadata(ctx, HookCommentPreModerate, userID, data, map[string]any{ + "result": result, + }) +} + +func (h *HookTriggerHelper) TriggerCommentModerated(ctx context.Context, userID uint, data *CommentModeratedHookData) { + h.manager.Trigger(ctx, HookCommentModerated, userID, data) +} diff --git a/internal/router/router.go b/internal/router/router.go index 3304034..f53b1ab 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -32,6 +32,7 @@ type Router struct { adminGroupHandler *handler.AdminGroupHandler adminDashboardHandler *handler.AdminDashboardHandler adminLogHandler *handler.AdminLogHandler + qrcodeHandler *handler.QRCodeHandler logService *service.LogService jwtService *service.JWTService casbinService service.CasbinService @@ -60,6 +61,7 @@ func New( adminGroupHandler *handler.AdminGroupHandler, adminDashboardHandler *handler.AdminDashboardHandler, adminLogHandler *handler.AdminLogHandler, + qrcodeHandler *handler.QRCodeHandler, logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, @@ -91,6 +93,7 @@ func New( adminGroupHandler: adminGroupHandler, adminDashboardHandler: adminDashboardHandler, adminLogHandler: adminLogHandler, + qrcodeHandler: qrcodeHandler, logService: logService, jwtService: jwtService, casbinService: casbinService, @@ -183,6 +186,24 @@ func (r *Router) setupRoutes() { users.GET("/:id/favorites", middleware.OptionalAuth(r.jwtService), r.postHandler.GetFavorites) } + // 二维码登录（公开） + if r.qrcodeHandler != nil { + qrcode := v1.Group("/auth/qrcode") + { + qrcode.GET("", r.qrcodeHandler.GetQRCode) + qrcode.GET("/events", r.qrcodeHandler.SSEEvents) + } + + // 二维码操作（需认证） + qrcodeAuth := v1.Group("/auth/qrcode") + qrcodeAuth.Use(authMiddleware) + { + qrcodeAuth.POST("/scan", r.qrcodeHandler.Scan) + qrcodeAuth.POST("/confirm", r.qrcodeHandler.Confirm) + qrcodeAuth.POST("/cancel", r.qrcodeHandler.Cancel) + } + } + // 认证路由（公开） authPublic := v1.Group("/auth") { diff --git a/internal/service/comment_service.go b/internal/service/comment_service.go index 0d167a3..32c1b09 100644 --- a/internal/service/comment_service.go +++ b/internal/service/comment_service.go @@ -2,19 +2,18 @@ package service import ( "context" - "errors" "fmt" "strings" "carrot_bbs/internal/cache" "carrot_bbs/internal/model" "carrot_bbs/internal/pkg/gorse" + "carrot_bbs/internal/pkg/hook" "carrot_bbs/internal/repository" "go.uber.org/zap" ) -// CommentService 评论服务 type CommentService struct { commentRepo *repository.CommentRepository postRepo *repository.PostRepository @@ -23,10 +22,10 @@ type CommentService struct { gorseClient gorse.Client postAIService *PostAIService logService *LogService + hookManager *hook.Manager } -// NewCommentService 创建评论服务 -func NewCommentService(commentRepo *repository.CommentRepository, postRepo *repository.PostRepository, systemMessageService SystemMessageService, gorseClient gorse.Client, postAIService *PostAIService, cacheBackend cache.Cache) *CommentService { +func NewCommentService(commentRepo *repository.CommentRepository, postRepo *repository.PostRepository, systemMessageService SystemMessageService, gorseClient gorse.Client, postAIService *PostAIService, cacheBackend cache.Cache, hookManager *hook.Manager) *CommentService { return &CommentService{ commentRepo: commentRepo, postRepo: postRepo, @@ -35,6 +34,7 @@ func NewCommentService(commentRepo *repository.CommentRepository, postRepo *repo gorseClient: gorseClient, postAIService: postAIService, logService: nil, + hookManager: hookManager, } } @@ -113,10 +113,9 @@ func (s *CommentService) reviewCommentAsync( parentUserID string, postOwnerID string, ) { - // 未启用AI时，直接通过审核并发送后续通知 - if s.postAIService == nil || !s.postAIService.IsEnabled() { + if s.hookManager == nil { if err := s.commentRepo.UpdateModerationStatus(commentID, model.CommentStatusPublished); err != nil { - zap.L().Warn("Failed to publish comment without AI moderation", + zap.L().Warn("Failed to publish comment without hook manager", zap.Error(err), ) return @@ -132,40 +131,38 @@ func (s *CommentService) reviewCommentAsync( return } - err := s.postAIService.ModerateComment(context.Background(), content, imageURLs) - if err != nil { - var rejectedErr *CommentModerationRejectedError - if errors.As(err, &rejectedErr) { - if delErr := s.commentRepo.Delete(commentID); delErr != nil { - zap.L().Warn("Failed to delete rejected comment", - zap.String("commentID", commentID), - zap.Error(delErr), - ) - } - s.notifyCommentModerationRejected(userID, rejectedErr.Reason) - return - } + var authorID uint + fmt.Sscanf(userID, "%d", &authorID) - // 审核服务异常时降级放行，避免评论长期pending - if updateErr := s.commentRepo.UpdateModerationStatus(commentID, model.CommentStatusPublished); updateErr != nil { - zap.L().Warn("Failed to publish comment after moderation error", + result := &hook.ModerationResult{} + s.hookManager.TriggerWithMetadata(context.Background(), hook.HookCommentPreModerate, authorID, &hook.CommentModerateHookData{ + CommentID: commentID, + PostID: postID, + Content: content, + Images: imageURLs, + AuthorID: authorID, + ParentID: parentID, + }, map[string]interface{}{ + "result": result, + }) + + s.hookManager.Trigger(context.Background(), hook.HookCommentModerated, authorID, &hook.CommentModeratedHookData{ + CommentID: commentID, + PostID: postID, + AuthorID: authorID, + Approved: result.Approved, + RejectReason: result.RejectReason, + ReviewedBy: result.ReviewedBy, + }) + + if !result.Approved { + if delErr := s.commentRepo.Delete(commentID); delErr != nil { + zap.L().Warn("Failed to delete rejected comment", zap.String("commentID", commentID), - zap.Error(updateErr), - ) - return - } - if statsErr := s.applyCommentPublishedStats(commentID); statsErr != nil { - zap.L().Warn("Failed to apply published stats for comment", - zap.String("commentID", commentID), - zap.Error(statsErr), + zap.Error(delErr), ) } - s.invalidatePostCaches(postID) - zap.L().Warn("Comment moderation failed, fallback publish comment", - zap.String("commentID", commentID), - zap.Error(err), - ) - s.afterCommentPublished(userID, postID, commentID, parentID, parentUserID, postOwnerID) + s.notifyCommentModerationRejected(userID, result.RejectReason) return } diff --git a/internal/service/post_service.go b/internal/service/post_service.go index e558198..29a9449 100644 --- a/internal/service/post_service.go +++ b/internal/service/post_service.go @@ -2,7 +2,6 @@ package service import ( "context" - "errors" "fmt" "log" "math/rand" @@ -12,6 +11,7 @@ import ( "carrot_bbs/internal/cache" "carrot_bbs/internal/model" "carrot_bbs/internal/pkg/gorse" + "carrot_bbs/internal/pkg/hook" "carrot_bbs/internal/repository" ) @@ -73,10 +73,10 @@ type postServiceImpl struct { postAIService *PostAIService txManager repository.TransactionManager logService *LogService + hookManager *hook.Manager } -// NewPostService 创建帖子服务 -func NewPostService(postRepo *repository.PostRepository, systemMessageService SystemMessageService, gorseClient gorse.Client, postAIService *PostAIService, cacheBackend cache.Cache, txManager repository.TransactionManager) PostService { +func NewPostService(postRepo *repository.PostRepository, systemMessageService SystemMessageService, gorseClient gorse.Client, postAIService *PostAIService, cacheBackend cache.Cache, txManager repository.TransactionManager, hookManager *hook.Manager) PostService { return &postServiceImpl{ postRepo: postRepo, systemMessageService: systemMessageService, @@ -85,6 +85,7 @@ func NewPostService(postRepo *repository.PostRepository, systemMessageService Sy postAIService: postAIService, txManager: txManager, logService: nil, + hookManager: hookManager, } } @@ -147,40 +148,48 @@ func (s *postServiceImpl) reviewPostAsync(postID, userID, title, content string, } }() - // 未启用AI时，直接发布 - if s.postAIService == nil || !s.postAIService.IsEnabled() { + if s.hookManager == nil { if err := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", "system"); err != nil { - log.Printf("[WARN] Failed to publish post without AI moderation: %v", err) + log.Printf("[WARN] Failed to publish post without hook manager: %v", err) } else { s.invalidatePostCaches(postID) } return } - err := s.postAIService.ModeratePost(context.Background(), title, content, images) - if err != nil { - var rejectedErr *PostModerationRejectedError - if errors.As(err, &rejectedErr) { - if updateErr := s.updateModerationStatusWithRetry(postID, model.PostStatusRejected, rejectedErr.UserMessage(), "ai"); updateErr != nil { - log.Printf("[WARN] Failed to reject post %s: %v", postID, updateErr) - } else { - s.invalidatePostCaches(postID) - } - s.notifyModerationRejected(userID, rejectedErr.Reason) - return - } + var authorID uint + fmt.Sscanf(userID, "%d", &authorID) - // 规则审核不可用时，降级为发布，避免长时间pending - if updateErr := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", "system"); updateErr != nil { - log.Printf("[WARN] Failed to publish post %s after moderation error: %v", postID, updateErr) + result := &hook.ModerationResult{} + s.hookManager.TriggerWithMetadata(context.Background(), hook.HookPostPreModerate, authorID, &hook.PostModerateHookData{ + PostID: postID, + Title: title, + Content: content, + Images: images, + AuthorID: authorID, + }, map[string]interface{}{ + "result": result, + }) + + s.hookManager.Trigger(context.Background(), hook.HookPostModerated, authorID, &hook.PostModeratedHookData{ + PostID: postID, + AuthorID: authorID, + Approved: result.Approved, + RejectReason: result.RejectReason, + ReviewedBy: result.ReviewedBy, + }) + + if !result.Approved { + if updateErr := s.updateModerationStatusWithRetry(postID, model.PostStatusRejected, result.RejectReason, result.ReviewedBy); updateErr != nil { + log.Printf("[WARN] Failed to reject post %s: %v", postID, updateErr) } else { s.invalidatePostCaches(postID) } - log.Printf("[WARN] Post moderation failed, fallback publish post=%s err=%v", postID, err) + s.notifyModerationRejected(userID, result.RejectReason) return } - if err := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", "ai"); err != nil { + if err := s.updateModerationStatusWithRetry(postID, model.PostStatusPublished, "", result.ReviewedBy); err != nil { log.Printf("[WARN] Failed to publish post %s: %v", postID, err) return } diff --git a/internal/service/qrcode_login_service.go b/internal/service/qrcode_login_service.go new file mode 100644 index 0000000..6dec665 --- /dev/null +++ b/internal/service/qrcode_login_service.go @@ -0,0 +1,291 @@ +package service + +import ( + "context" + "fmt" + "time" + + "github.com/google/uuid" + + "carrot_bbs/internal/model" + "carrot_bbs/internal/pkg/redis" + "carrot_bbs/internal/pkg/sse" +) + +// Lua 脚本：原子性地检查状态并更新为 scanned +const scanScript = ` + local key = KEYS[1] + local status = redis.call('HGET', key, 'status') + if status == 'pending' then + redis.call('HMSET', key, 'status', 'scanned', 'user_id', ARGV[1]) + return 1 + elseif status == 'scanned' then + return 2 + else + return 0 + end +` + +const ( + qrcodeSessionPrefix = "qrcode:session:" + qrcodeSessionTTL = 5 * time.Minute +) + +// QRCodeStatus 二维码状态 +type QRCodeStatus string + +const ( + QRCodeStatusPending QRCodeStatus = "pending" + QRCodeStatusScanned QRCodeStatus = "scanned" + QRCodeStatusConfirmed QRCodeStatus = "confirmed" + QRCodeStatusCancelled QRCodeStatus = "cancelled" + QRCodeStatusExpired QRCodeStatus = "expired" +) + +// QRCodeSession 二维码会话 +type QRCodeSession struct { + SessionID string `json:"session_id"` + Status QRCodeStatus `json:"status"` + UserID string `json:"user_id"` + CreatedAt int64 `json:"created_at"` + ExpiresAt int64 `json:"expires_at"` +} + +// QRCodeLoginService 二维码登录服务 +type QRCodeLoginService struct { + redis *redis.Client + sseHub *sse.Hub + jwtService *JWTService + userService UserService + activityService UserActivityService + logService *LogService +} + +// NewQRCodeLoginService 创建二维码登录服务 +func NewQRCodeLoginService(redis *redis.Client, sseHub *sse.Hub, jwtService *JWTService, userService UserService, activityService UserActivityService, logService *LogService) *QRCodeLoginService { + return &QRCodeLoginService{ + redis: redis, + sseHub: sseHub, + jwtService: jwtService, + userService: userService, + activityService: activityService, + logService: logService, + } +} + +// CreateSession 创建二维码会话 +func (s *QRCodeLoginService) CreateSession(ctx context.Context) (*QRCodeSession, error) { + sessionID := uuid.New().String() + now := time.Now() + expiresAt := now.Add(qrcodeSessionTTL) + + session := &QRCodeSession{ + SessionID: sessionID, + Status: QRCodeStatusPending, + CreatedAt: now.Unix(), + ExpiresAt: expiresAt.Unix(), + } + + key := qrcodeSessionPrefix + sessionID + data := map[string]interface{}{ + "status": string(session.Status), + "user_id": "", + "created_at": session.CreatedAt, + "expires_at": session.ExpiresAt, + } + + if err := s.redis.HMSet(ctx, key, data); err != nil { + return nil, fmt.Errorf("failed to create session: %w", err) + } + + if _, err := s.redis.Expire(ctx, key, qrcodeSessionTTL); err != nil { + return nil, fmt.Errorf("failed to set session ttl: %w", err) + } + + return session, nil +} + +// GetSession 获取会话 +func (s *QRCodeLoginService) GetSession(ctx context.Context, sessionID string) (*QRCodeSession, error) { + key := qrcodeSessionPrefix + sessionID + data, err := s.redis.HGetAll(ctx, key) + if err != nil { + return nil, fmt.Errorf("failed to get session: %w", err) + } + + if len(data) == 0 { + return nil, fmt.Errorf("session not found or expired") + } + + session := &QRCodeSession{ + SessionID: sessionID, + Status: QRCodeStatus(data["status"]), + UserID: data["user_id"], + } + + // 解析时间戳 + if v := data["created_at"]; v != "" { + fmt.Sscanf(v, "%d", &session.CreatedAt) + } + if v := data["expires_at"]; v != "" { + fmt.Sscanf(v, "%d", &session.ExpiresAt) + } + + // 检查是否过期 + if time.Now().Unix() > session.ExpiresAt { + session.Status = QRCodeStatusExpired + } + + return session, nil +} + +// Scan 扫描二维码 +func (s *QRCodeLoginService) Scan(ctx context.Context, sessionID, userID string) error { + key := qrcodeSessionPrefix + sessionID + + // 使用 Lua 脚本实现原子性检查和更新，避免竞态条件 + result, err := s.redis.GetClient().Eval(ctx, scanScript, []string{key}, userID).Int() + if err != nil { + return fmt.Errorf("failed to scan: %w", err) + } + + switch result { + case 0: + return fmt.Errorf("session not found or expired") + case 2: + return fmt.Errorf("qrcode already scanned") + } + + // 获取用户信息用于推送 + user, err := s.userService.GetUserByID(ctx, userID) + if err != nil { + return fmt.Errorf("failed to get user info: %w", err) + } + + // 推送扫码事件 + payload := map[string]any{ + "user": map[string]any{ + "id": user.ID, + "nickname": user.Nickname, + "avatar": user.Avatar, + }, + } + s.sseHub.PublishToUser(sessionID, "scanned", payload) + + return nil +} + +// Confirm 确认登录 +func (s *QRCodeLoginService) Confirm(ctx context.Context, sessionID, userID string, ip, userAgent string) (*LoginResponse, error) { + session, err := s.GetSession(ctx, sessionID) + if err != nil { + return nil, err + } + + if session.Status != QRCodeStatusScanned { + return nil, fmt.Errorf("invalid session status") + } + + if session.UserID != userID { + return nil, fmt.Errorf("unauthorized") + } + + // 获取用户信息 + user, err := s.userService.GetUserByID(ctx, userID) + if err != nil { + return nil, fmt.Errorf("user not found: %w", err) + } + + // 复用JWT生成逻辑 + accessToken, err := s.jwtService.GenerateAccessToken(user.ID, user.Username) + if err != nil { + return nil, fmt.Errorf("failed to generate access token: %w", err) + } + + refreshToken, err := s.jwtService.GenerateRefreshToken(user.ID, user.Username) + if err != nil { + return nil, fmt.Errorf("failed to generate refresh token: %w", err) + } + + // 更新状态 + key := qrcodeSessionPrefix + sessionID + if err := s.redis.HSet(ctx, key, "status", string(QRCodeStatusConfirmed)); err != nil { + return nil, fmt.Errorf("failed to update status: %w", err) + } + + // 复用活跃记录 + if s.activityService != nil { + go func() { + s.activityService.RecordUserActive(context.Background(), userID, model.LoginTypeLogin, ip, userAgent) + }() + } + + // 复用登录日志 + if s.logService != nil { + go func() { + s.logService.LoginLog.RecordLogin(&model.LoginLog{ + UserID: user.ID, + UserName: user.Username, + NickName: user.Nickname, + LoginType: string(model.LoginTypeQRCode), + Event: string(model.LoginEventLogin), + IP: ip, + UserAgent: userAgent, + Result: string(model.LoginResultSuccess), + }) + }() + } + + // 推送确认事件 + response := &LoginResponse{ + Token: accessToken, + RefreshToken: refreshToken, + User: user, + } + s.sseHub.PublishToUser(sessionID, "confirmed", response) + + return response, nil +} + +// LoginResponse 登录响应 +type LoginResponse struct { + Token string `json:"token"` + RefreshToken string `json:"refresh_token"` + User *model.User `json:"user"` +} + +// Cancel 取消登录 +func (s *QRCodeLoginService) Cancel(ctx context.Context, sessionID, userID string) error { + session, err := s.GetSession(ctx, sessionID) + if err != nil { + return err + } + + if session.Status != QRCodeStatusScanned { + return fmt.Errorf("invalid session status") + } + + if session.UserID != userID { + return fmt.Errorf("unauthorized") + } + + key := qrcodeSessionPrefix + sessionID + if err := s.redis.HSet(ctx, key, "status", string(QRCodeStatusCancelled)); err != nil { + return fmt.Errorf("failed to update status: %w", err) + } + + // 推送取消事件 + s.sseHub.PublishToUser(sessionID, "cancelled", map[string]interface{}{}) + + return nil +} + +// GetSSEHub 获取SSE Hub +func (s *QRCodeLoginService) GetSSEHub() *sse.Hub { + return s.sseHub +} + +// GetUserService 获取用户服务 +func (s *QRCodeLoginService) GetUserService() UserService { + return s.userService +} diff --git a/internal/wire/handler.go b/internal/wire/handler.go index 346b9c0..cb433ef 100644 --- a/internal/wire/handler.go +++ b/internal/wire/handler.go @@ -28,6 +28,7 @@ var HandlerSet = wire.NewSet( handler.NewAdminGroupHandler, handler.NewAdminDashboardHandler, handler.NewAdminLogHandler, + handler.NewQRCodeHandler, // 需要特殊处理的 Handler ProvideUserHandler, diff --git a/internal/wire/infrastructure.go b/internal/wire/infrastructure.go index 0c37cba..797fe41 100644 --- a/internal/wire/infrastructure.go +++ b/internal/wire/infrastructure.go @@ -9,11 +9,13 @@ import ( "carrot_bbs/internal/pkg/crypto" "carrot_bbs/internal/pkg/email" "carrot_bbs/internal/pkg/gorse" + "carrot_bbs/internal/pkg/hook" "carrot_bbs/internal/pkg/openai" "carrot_bbs/internal/pkg/redis" "carrot_bbs/internal/pkg/s3" "carrot_bbs/internal/pkg/sse" "carrot_bbs/internal/repository" + "carrot_bbs/internal/service" "github.com/google/wire" "go.uber.org/zap" @@ -38,6 +40,11 @@ var InfrastructureSet = wire.NewSet( ProvideRedisClient, ProvideCache, + // 钩子系统 + ProvideHookManager, + ProvideBuiltinHooks, + ProvideModerationHooks, + // SSE Hub ProvideSSEHub, @@ -88,7 +95,18 @@ func ProvideCache(cfg *config.Config, redisClient *redis.Client) cache.Cache { DisableFlushDB: cfg.Cache.DisableFlushDB, }) - // 直接创建 RedisCache 实例，不依赖全局变量 + if cfg.Cache.Local.Enabled { + layeredCache := cache.NewLayeredCache(redisClient, &cache.LayeredCacheOptions{ + LocalSize: cfg.Cache.Local.Size, + LocalBuckets: cfg.Cache.Local.Buckets, + LocalDefaultTTL: time.Duration(cfg.Cache.Local.DefaultTTL) * time.Second, + KeyPrefix: cfg.Cache.KeyPrefix, + Enabled: cfg.Cache.Enabled, + }) + zap.L().Info("Initialized layered cache (local LRU + Redis)") + return layeredCache + } + if redisClient == nil { zap.L().Warn("Redis client is nil, cache will not be available") return nil @@ -99,6 +117,47 @@ func ProvideCache(cfg *config.Config, redisClient *redis.Client) cache.Cache { return cacheInstance } +// ProvideHookManager 提供钩子管理器 +func ProvideHookManager() *hook.Manager { + return hook.NewManager(&hook.ManagerOptions{ + MaxConcurrentAsync: 100, + Enabled: true, + }) +} + +// ProvideBuiltinHooks 注册内置钩子 +func ProvideBuiltinHooks(manager *hook.Manager) *hook.BuiltinHooks { + builtin := hook.NewBuiltinHooks(manager) + builtin.RegisterAll() + zap.L().Info("Registered builtin hooks") + return builtin +} + +// ProvideModerationHooks 注册审核钩子 +func ProvideModerationHooks( + manager *hook.Manager, + postAIService *service.PostAIService, + postRepo *repository.PostRepository, + commentRepo *repository.CommentRepository, + cfg *config.Config, +) *hook.ModerationHooks { + strictMode := cfg.OpenAI.StrictModeration + + moderationHooks := hook.NewModerationHooks( + postAIService, + nil, + postRepo, + commentRepo, + strictMode, + "***", + ) + moderationHooks.RegisterAll(manager) + zap.L().Info("Registered moderation hooks", + zap.Bool("strict_mode", strictMode), + ) + return moderationHooks +} + // ProvideSSEHub 提供 SSE Hub func ProvideSSEHub() *sse.Hub { return sse.NewHub() diff --git a/internal/wire/service.go b/internal/wire/service.go index a319ee5..a4126f3 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -8,7 +8,9 @@ import ( "carrot_bbs/internal/grpc/runner" "carrot_bbs/internal/pkg/email" "carrot_bbs/internal/pkg/gorse" + "carrot_bbs/internal/pkg/hook" "carrot_bbs/internal/pkg/openai" + "carrot_bbs/internal/pkg/redis" "carrot_bbs/internal/pkg/s3" "carrot_bbs/internal/pkg/sse" "carrot_bbs/internal/repository" @@ -50,6 +52,7 @@ var ServiceSet = wire.NewSet( ProvideAdminCommentService, ProvideAdminGroupService, ProvideAdminDashboardService, + ProvideQRCodeLoginService, // 日志服务 ProvideAsyncLogManager, @@ -100,7 +103,6 @@ func ProvideSystemMessageService( return service.NewSystemMessageService(notifyRepo, pushService, userRepo, postRepo, cacheBackend) } -// ProvidePostService 提供帖子服务 func ProvidePostService( postRepo *repository.PostRepository, systemMessageService service.SystemMessageService, @@ -108,11 +110,11 @@ func ProvidePostService( postAIService *service.PostAIService, cacheBackend cache.Cache, txManager repository.TransactionManager, + hookManager *hook.Manager, ) service.PostService { - return service.NewPostService(postRepo, systemMessageService, gorseClient, postAIService, cacheBackend, txManager) + return service.NewPostService(postRepo, systemMessageService, gorseClient, postAIService, cacheBackend, txManager, hookManager) } -// ProvideCommentService 提供评论服务 func ProvideCommentService( commentRepo *repository.CommentRepository, postRepo *repository.PostRepository, @@ -120,8 +122,9 @@ func ProvideCommentService( gorseClient gorse.Client, postAIService *service.PostAIService, cacheBackend cache.Cache, + hookManager *hook.Manager, ) *service.CommentService { - return service.NewCommentService(commentRepo, postRepo, systemMessageService, gorseClient, postAIService, cacheBackend) + return service.NewCommentService(commentRepo, postRepo, systemMessageService, gorseClient, postAIService, cacheBackend, hookManager) } // ProvideMessageService 提供消息服务 @@ -343,6 +346,18 @@ func ProvideLogService( return service.NewLogService(operationLogService, loginLogService, dataChangeLogService) } +// ProvideQRCodeLoginService 提供二维码登录服务 +func ProvideQRCodeLoginService( + redisClient *redis.Client, + sseHub *sse.Hub, + jwtService *service.JWTService, + userService service.UserService, + activityService service.UserActivityService, + logService *service.LogService, +) *service.QRCodeLoginService { + return service.NewQRCodeLoginService(redisClient, sseHub, jwtService, userService, activityService, logService) +} + // parseDuration 解析时长字符串 func parseDuration(s string) (time.Duration, error) { return time.ParseDuration(s)