From a6df8c91314a934ae63cf3cbd6d7dfbe36a615a0 Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sat, 20 Jun 2026 23:51:45 +0800 Subject: [PATCH 01/10] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E9=99=90?= =?UTF-8?q?=E6=B5=81=E9=85=8D=E7=BD=AE=E5=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Config 结构体新增 RateLimit 字段 - 新增 RateLimitConfig 和 BucketConfig 结构体定义 - config.yaml 新增 ratelimit 配置段(默认关闭) - 设置默认值:query(10/0.2)、login(5/0.1)、register(3/0.05) --- backend/config.yaml | 15 ++++++++++++ backend/internal/config/config.go | 40 ++++++++++++++++++++++++------- 2 files changed, 47 insertions(+), 8 deletions(-) diff --git a/backend/config.yaml b/backend/config.yaml index 0aafbf6..f31a9cb 100644 --- a/backend/config.yaml +++ b/backend/config.yaml @@ -60,6 +60,21 @@ auth: access_ttl: 120 # Access Token 过期时间(分钟) refresh_ttl: 10080 # Refresh Token 过期时间(分钟),7 天 +ratelimit: + enabled: false # 是否启用限流 + # WebSocket query 消息限流(核心,控制 AI 成本) + query: + capacity: 10 # 突发容量:允许连续发 10 个 query + rate: 0.2 # 填充速率:每 5 秒补充 1 个令牌 + # REST API 登录限流(防暴力破解) + login: + capacity: 5 # 突发容量:允许连续 5 次登录尝试 + rate: 0.1 # 填充速率:每 10 秒补充 1 次 + # REST API 注册限流 + register: + capacity: 3 # 突发容量:允许连续 3 次注册 + rate: 0.05 # 填充速率:每 20 秒补充 1 次 + log: level: info # debug / info / warn / error format: console # console / json diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 2d67907..dd62719 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -10,14 +10,15 @@ import ( // Config 应用配置。 type Config struct { - App AppConfig `mapstructure:"app"` - Server ServerConfig `mapstructure:"server"` - Session SessionConfig `mapstructure:"session"` - Redis RedisConfig `mapstructure:"redis"` - AI AIConfig `mapstructure:"ai"` - Storage StorageConfig `mapstructure:"storage"` - Log LogConfig `mapstructure:"log"` - Auth AuthConfig `mapstructure:"auth"` + App AppConfig `mapstructure:"app"` + Server ServerConfig `mapstructure:"server"` + Session SessionConfig `mapstructure:"session"` + Redis RedisConfig `mapstructure:"redis"` + AI AIConfig `mapstructure:"ai"` + Storage StorageConfig `mapstructure:"storage"` + Log LogConfig `mapstructure:"log"` + Auth AuthConfig `mapstructure:"auth"` + RateLimit RateLimitConfig `mapstructure:"ratelimit"` } // SessionConfig 会话管理配置。 @@ -120,6 +121,20 @@ type AuthConfig struct { RefreshTTL int `mapstructure:"refresh_ttl"` // Refresh Token 过期时间(分钟),默认 10080(7天) } +// RateLimitConfig 限流配置。 +type RateLimitConfig struct { + Enabled bool `mapstructure:"enabled"` + Query BucketConfig `mapstructure:"query"` + Login BucketConfig `mapstructure:"login"` + Register BucketConfig `mapstructure:"register"` +} + +// BucketConfig 令牌桶配置。 +type BucketConfig struct { + Capacity int `mapstructure:"capacity"` // 桶容量(突发上限) + Rate float64 `mapstructure:"rate"` // 每秒填充令牌数 +} + // Load 加载配置。优先级:环境变量 > config.{env}.yaml > config.yaml > 默认值。 // workDir 为项目根目录或 backend 目录,用于定位 .env 和 config.yaml。 func Load(workDir string) (*Config, error) { @@ -218,6 +233,15 @@ func setDefaults(v *viper.Viper) { // log v.SetDefault("log.level", "info") v.SetDefault("log.format", "console") + + // ratelimit + v.SetDefault("ratelimit.enabled", false) + v.SetDefault("ratelimit.query.capacity", 10) + v.SetDefault("ratelimit.query.rate", 0.2) + v.SetDefault("ratelimit.login.capacity", 5) + v.SetDefault("ratelimit.login.rate", 0.1) + v.SetDefault("ratelimit.register.capacity", 3) + v.SetDefault("ratelimit.register.rate", 0.05) } // bindEnvVars 显式绑定敏感信息环境变量。 From 3b6226394b495f861b11676bc965571f9a1e1d08 Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sat, 20 Jun 2026 23:53:36 +0800 Subject: [PATCH 02/10] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E5=86=85?= =?UTF-8?q?=E5=AD=98=E4=BB=A4=E7=89=8C=E6=A1=B6=E9=99=90=E6=B5=81=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Limiter 接口定义(Allow + Stop 方法) - TokenBucket 实现(容量、填充速率、并发安全) - MemoryLimiter 管理多用户令牌桶 - 后台 goroutine 定期清理不活跃桶(10 分钟) - 完整单元测试覆盖(11 个测试用例,全部通过) - 边界情况处理(rate=0、capacity=0、并发安全) --- backend/internal/ratelimit/bucket.go | 172 ++++++++++++++++++ backend/internal/ratelimit/bucket_test.go | 203 ++++++++++++++++++++++ backend/internal/ratelimit/limiter.go | 17 ++ 3 files changed, 392 insertions(+) create mode 100644 backend/internal/ratelimit/bucket.go create mode 100644 backend/internal/ratelimit/bucket_test.go create mode 100644 backend/internal/ratelimit/limiter.go diff --git a/backend/internal/ratelimit/bucket.go b/backend/internal/ratelimit/bucket.go new file mode 100644 index 0000000..308336f --- /dev/null +++ b/backend/internal/ratelimit/bucket.go @@ -0,0 +1,172 @@ +package ratelimit + +import ( + "context" + "sync" + "time" + + "github.com/hhs/camtalk/internal/config" +) + +// TokenBucket 内存令牌桶,适用于单实例部署。 +type TokenBucket struct { + capacity int // 桶容量 + rate float64 // 每秒填充令牌数 + tokens float64 // 当前令牌数 + lastRefill time.Time // 上次填充时间 + mu sync.Mutex +} + +// newTokenBucket 创建令牌桶。 +func newTokenBucket(capacity int, rate float64) *TokenBucket { + return &TokenBucket{ + capacity: capacity, + rate: rate, + tokens: float64(capacity), // 初始满桶 + lastRefill: time.Now(), + } +} + +// allow 尝试消耗一个令牌。 +func (b *TokenBucket) allow() (bool, time.Duration) { + b.mu.Lock() + defer b.mu.Unlock() + + now := time.Now() + elapsed := now.Sub(b.lastRefill).Seconds() + + // 补充令牌 + newTokens := elapsed * b.rate + b.tokens = min(float64(b.capacity), b.tokens+newTokens) + b.lastRefill = now + + // 尝试消耗一个令牌 + if b.tokens >= 1 { + b.tokens -= 1 + return true, 0 + } + + // 计算需要等待的时间 + if b.rate == 0 { + // rate=0 时永远无法补充令牌 + return false, 24 * time.Hour // 返回一个很大的值 + } + retryAfter := time.Duration((1-b.tokens)/b.rate*1000) * time.Millisecond + return false, retryAfter +} + +// MemoryLimiter 管理多个用户的令牌桶。 +type MemoryLimiter struct { + buckets map[string]*TokenBucket + config config.RateLimitConfig + mu sync.RWMutex + stopOnce sync.Once + done chan struct{} +} + +// NewMemoryLimiter 创建内存限流器。 +func NewMemoryLimiter(cfg config.RateLimitConfig) *MemoryLimiter { + limiter := &MemoryLimiter{ + buckets: make(map[string]*TokenBucket), + config: cfg, + done: make(chan struct{}), + } + + // 启动后台清理 goroutine + go limiter.cleanup() + + return limiter +} + +// Allow 实现 Limiter 接口。 +func (l *MemoryLimiter) Allow(ctx context.Context, key string) (bool, time.Duration) { + bucket := l.getOrCreateBucket(key) + return bucket.allow() +} + +// Stop 实现 Limiter 接口。 +func (l *MemoryLimiter) Stop() { + l.stopOnce.Do(func() { + close(l.done) + }) +} + +// getOrCreateBucket 获取或创建令牌桶。 +func (l *MemoryLimiter) getOrCreateBucket(key string) *TokenBucket { + // 先尝试读锁 + l.mu.RLock() + bucket, exists := l.buckets[key] + l.mu.RUnlock() + + if exists { + return bucket + } + + // 需要创建新桶,升级为写锁 + l.mu.Lock() + defer l.mu.Unlock() + + // 双重检查(可能其他 goroutine 已创建) + bucket, exists = l.buckets[key] + if exists { + return bucket + } + + // 根据 key 确定配置(简化版:假设 key 格式为 "userID:action") + cfg := l.getBucketConfig(key) + bucket = newTokenBucket(cfg.Capacity, cfg.Rate) + l.buckets[key] = bucket + + return bucket +} + +// getBucketConfig 根据 key 获取桶配置。 +func (l *MemoryLimiter) getBucketConfig(key string) config.BucketConfig { + // 简化实现:从 key 后缀判断动作类型 + // 实际使用时调用方会传递正确的 key + // 默认使用 query 配置 + return l.config.Query +} + +// cleanup 定期清理不活跃的桶。 +func (l *MemoryLimiter) cleanup() { + ticker := time.NewTicker(10 * time.Minute) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + l.removeInactiveBuckets() + case <-l.done: + return + } + } +} + +// removeInactiveBuckets 移除超过 10 分钟无活动的桶。 +func (l *MemoryLimiter) removeInactiveBuckets() { + l.mu.Lock() + defer l.mu.Unlock() + + now := time.Now() + for key, bucket := range l.buckets { + bucket.mu.Lock() + inactive := now.Sub(bucket.lastRefill) > 10*time.Minute + bucket.mu.Unlock() + + if inactive { + delete(l.buckets, key) + } + } +} + +// min 返回两个 float64 中的较小值。 +func min(a, b float64) float64 { + if a < b { + return a + } + return b +} + +// 编译期接口检查 +var _ Limiter = (*MemoryLimiter)(nil) diff --git a/backend/internal/ratelimit/bucket_test.go b/backend/internal/ratelimit/bucket_test.go new file mode 100644 index 0000000..ebb19e9 --- /dev/null +++ b/backend/internal/ratelimit/bucket_test.go @@ -0,0 +1,203 @@ +package ratelimit + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/hhs/camtalk/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestTokenBucket_Allow_FirstRequest(t *testing.T) { + bucket := newTokenBucket(5, 0.2) + + allowed, retryAfter := bucket.allow() + + assert.True(t, allowed) + assert.Equal(t, time.Duration(0), retryAfter) +} + +func TestTokenBucket_Allow_ConsumeUntilEmpty(t *testing.T) { + bucket := newTokenBucket(3, 0.2) + + // 连续消耗 3 个令牌 + for i := 0; i < 3; i++ { + allowed, _ := bucket.allow() + assert.True(t, allowed, "request %d should be allowed", i+1) + } + + // 第 4 个请求应被拒绝 + allowed, retryAfter := bucket.allow() + assert.False(t, allowed) + assert.Greater(t, retryAfter, time.Duration(0)) +} + +func TestTokenBucket_Allow_RetryAfterCorrect(t *testing.T) { + bucket := newTokenBucket(1, 1.0) // 每秒 1 个令牌 + + // 消耗唯一的令牌 + allowed, _ := bucket.allow() + require.True(t, allowed) + + // 立即再次请求应被拒绝 + allowed, retryAfter := bucket.allow() + assert.False(t, allowed) + // retryAfter 应约为 1 秒(允许一定误差) + assert.InDelta(t, 1000, retryAfter.Milliseconds(), 100) +} + +func TestTokenBucket_Allow_RefillAfterWait(t *testing.T) { + bucket := newTokenBucket(2, 10.0) // 每秒 10 个令牌(每 100ms 一个) + + // 消耗 2 个令牌 + bucket.allow() + bucket.allow() + + // 等待 150ms,应补充至少 1 个令牌 + time.Sleep(150 * time.Millisecond) + + allowed, _ := bucket.allow() + assert.True(t, allowed) +} + +func TestTokenBucket_Allow_CapacityLimit(t *testing.T) { + bucket := newTokenBucket(3, 1.0) + + // 等待足够长时间让桶"溢出" + time.Sleep(100 * time.Millisecond) + + // 但最多只能消耗 capacity 个令牌 + for i := 0; i < 3; i++ { + allowed, _ := bucket.allow() + assert.True(t, allowed, "request %d should be allowed", i+1) + } + + // 第 4 个应被拒绝 + allowed, _ := bucket.allow() + assert.False(t, allowed) +} + +func TestTokenBucket_Allow_ConcurrentSafe(t *testing.T) { + bucket := newTokenBucket(100, 10.0) + var wg sync.WaitGroup + successCount := 0 + var mu sync.Mutex + + // 100 个并发请求 + for i := 0; i < 100; i++ { + wg.Add(1) + go func() { + defer wg.Done() + allowed, _ := bucket.allow() + if allowed { + mu.Lock() + successCount++ + mu.Unlock() + } + }() + } + + wg.Wait() + + // 应该正好 100 个成功(桶容量为 100) + assert.Equal(t, 100, successCount) +} + +func TestTokenBucket_Allow_ZeroCapacity(t *testing.T) { + bucket := newTokenBucket(0, 1.0) + + allowed, retryAfter := bucket.allow() + assert.False(t, allowed) + assert.Greater(t, retryAfter, time.Duration(0)) +} + +func TestTokenBucket_Allow_ZeroRate(t *testing.T) { + bucket := newTokenBucket(1, 0.0) + + // 第一个通过 + allowed, _ := bucket.allow() + assert.True(t, allowed) + + // 第二个被拒绝,且 retryAfter 应为无限大(实际上会很大) + allowed, retryAfter := bucket.allow() + assert.False(t, allowed) + // rate=0 时,retryAfter 理论上无限大,实际会是一个很大的值 + assert.Greater(t, retryAfter, 1*time.Hour) +} + +func TestMemoryLimiter_Allow_DifferentKeys(t *testing.T) { + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 2, Rate: 1.0}, + } + limiter := NewMemoryLimiter(cfg) + defer limiter.Stop() + + ctx := context.Background() + + // user1 消耗 2 个令牌 + allowed, _ := limiter.Allow(ctx, "user1:query") + assert.True(t, allowed) + allowed, _ = limiter.Allow(ctx, "user1:query") + assert.True(t, allowed) + + // user1 第 3 个被拒绝 + allowed, _ = limiter.Allow(ctx, "user1:query") + assert.False(t, allowed) + + // user2 应该不受影响 + allowed, _ = limiter.Allow(ctx, "user2:query") + assert.True(t, allowed) +} + +func TestMemoryLimiter_Cleanup(t *testing.T) { + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 1, Rate: 1.0}, + } + limiter := NewMemoryLimiter(cfg) + defer limiter.Stop() + + ctx := context.Background() + + // 创建一个桶 + limiter.Allow(ctx, "user1:query") + + // 验证桶已创建 + limiter.mu.RLock() + initialCount := len(limiter.buckets) + limiter.mu.RUnlock() + assert.Equal(t, 1, initialCount) + + // 手动触发清理(模拟 10 分钟后) + limiter.mu.Lock() + for _, bucket := range limiter.buckets { + bucket.mu.Lock() + bucket.lastRefill = time.Now().Add(-11 * time.Minute) + bucket.mu.Unlock() + } + limiter.mu.Unlock() + + limiter.removeInactiveBuckets() + + // 验证桶已被清理 + limiter.mu.RLock() + finalCount := len(limiter.buckets) + limiter.mu.RUnlock() + assert.Equal(t, 0, finalCount) +} + +func TestMemoryLimiter_Stop(t *testing.T) { + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 1, Rate: 1.0}, + } + limiter := NewMemoryLimiter(cfg) + + // 多次调用 Stop 不应 panic + limiter.Stop() + limiter.Stop() +} diff --git a/backend/internal/ratelimit/limiter.go b/backend/internal/ratelimit/limiter.go new file mode 100644 index 0000000..5fa4e3c --- /dev/null +++ b/backend/internal/ratelimit/limiter.go @@ -0,0 +1,17 @@ +package ratelimit + +import ( + "context" + "time" +) + +// Limiter 速率限制器接口。 +type Limiter interface { + // Allow 判断 key 是否允许执行一次操作。 + // key 通常为 "userID:action" 格式。 + // 返回 (allowed, retryAfter)。retryAfter 表示需要等待的时间。 + Allow(ctx context.Context, key string) (bool, time.Duration) + + // Stop 停止限流器,清理资源(如后台 goroutine)。 + Stop() +} From b74fb3564d833cae14d2b7360033f4716e6c8fc9 Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sat, 20 Jun 2026 23:55:37 +0800 Subject: [PATCH 03/10] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=20Redis=20?= =?UTF-8?q?=E4=BB=A4=E7=89=8C=E6=A1=B6=E9=99=90=E6=B5=81=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RedisLimiter 基于 Lua 脚本保证原子性 - Lua 脚本实现完整令牌桶算法(填充、消耗、TTL) - fail-open 策略:Redis 故障时允许请求通过 - FormatKey 辅助函数格式化限流 key - 完整单元测试(10 个测试用例,使用 miniredis) - 测试覆盖:首次请求、耗尽、不同用户、补充、容量上限、零速率、TTL、故障降级 --- backend/go.mod | 2 + backend/go.sum | 4 + backend/internal/ratelimit/redis_bucket.go | 128 ++++++++++ .../internal/ratelimit/redis_bucket_test.go | 228 ++++++++++++++++++ 4 files changed, 362 insertions(+) create mode 100644 backend/internal/ratelimit/redis_bucket.go create mode 100644 backend/internal/ratelimit/redis_bucket_test.go diff --git a/backend/go.mod b/backend/go.mod index 6f652f0..e9b487e 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -19,6 +19,7 @@ require ( ) require ( + github.com/alicebob/miniredis/v2 v2.38.0 // indirect github.com/bahlo/generic-list-go v0.2.0 // indirect github.com/buger/jsonparser v1.1.1 // indirect github.com/bytedance/gopkg v0.1.3 // indirect @@ -68,6 +69,7 @@ require ( github.com/ugorji/go/codec v1.2.12 // indirect github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect github.com/yargevad/filepathx v1.0.0 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.10.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect diff --git a/backend/go.sum b/backend/go.sum index 9d58137..93d4a44 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -1,4 +1,6 @@ github.com/airbrake/gobrake v3.6.1+incompatible/go.mod h1:wM4gu3Cn0W0K7GUuVWnlXZU11AGBXMILnrdOU8Kn00o= +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/bahlo/generic-list-go v0.2.0 h1:5sz/EEAK+ls5wF+NeqDpk5+iNdMDXrh3z3nPnH1Wvgk= github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xWbdbCW3pNTGyYg= github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA= @@ -190,6 +192,8 @@ github.com/x-cray/logrus-prefixed-formatter v0.5.2 h1:00txxvfBM9muc0jiLIEAkAcIMJ github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE= github.com/yargevad/filepathx v1.0.0 h1:SYcT+N3tYGi+NvazubCNlvgIPbzAk7i7y2dwg3I5FYc= github.com/yargevad/filepathx v1.0.0/go.mod h1:BprfX/gpYNJHJfc35GjRRpVcwWXS89gGulUIU5tK3tA= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= diff --git a/backend/internal/ratelimit/redis_bucket.go b/backend/internal/ratelimit/redis_bucket.go new file mode 100644 index 0000000..eb1e606 --- /dev/null +++ b/backend/internal/ratelimit/redis_bucket.go @@ -0,0 +1,128 @@ +package ratelimit + +import ( + "context" + "fmt" + "strconv" + "time" + + "github.com/hhs/camtalk/internal/config" + "github.com/redis/go-redis/v9" +) + +// luaScript 是 Redis 令牌桶算法的 Lua 脚本。 +// 保证原子性:读取-计算-回写在一个事务中完成。 +const luaScript = ` +-- KEYS[1] = 限流 key +-- ARGV[1] = capacity(桶容量) +-- ARGV[2] = rate(每秒填充数) +-- ARGV[3] = now(当前时间戳,秒,浮点) +-- ARGV[4] = ttl(key 过期时间,秒) + +local key = KEYS[1] +local capacity = tonumber(ARGV[1]) +local rate = tonumber(ARGV[2]) +local now = tonumber(ARGV[3]) +local ttl = tonumber(ARGV[4]) + +local data = redis.call('HMGET', key, 'tokens', 'last_refill') +local tokens = tonumber(data[1]) or capacity +local last_refill = tonumber(data[2]) or now + +-- 计算新令牌 +local elapsed = math.max(0, now - last_refill) +tokens = math.min(capacity, tokens + elapsed * rate) + +local allowed = 0 +local retry_after = 0 + +if tokens >= 1 then + tokens = tokens - 1 + allowed = 1 +else + if rate == 0 then + retry_after = 86400 -- 24小时 + else + retry_after = (1 - tokens) / rate + end +end + +-- 回写状态 +redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now) +redis.call('EXPIRE', key, ttl) + +return {allowed, tostring(retry_after)} +` + +// RedisLimiter Redis 令牌桶限流器。 +type RedisLimiter struct { + client *redis.Client + config config.RateLimitConfig + script *redis.Script +} + +// NewRedisLimiter 创建 Redis 限流器。 +func NewRedisLimiter(client *redis.Client, cfg config.RateLimitConfig) *RedisLimiter { + return &RedisLimiter{ + client: client, + config: cfg, + script: redis.NewScript(luaScript), + } +} + +// Allow 实现 Limiter 接口。 +func (l *RedisLimiter) Allow(ctx context.Context, key string) (bool, time.Duration) { + cfg := l.getBucketConfig(key) + + now := float64(time.Now().UnixNano()) / 1e9 // 秒,浮点 + ttl := 600 // key 过期时间 10 分钟 + + result, err := l.script.Run(ctx, l.client, []string{key}, + cfg.Capacity, cfg.Rate, now, ttl).Result() + + if err != nil { + // Redis 错误时降级:允许请求(fail-open 策略) + return true, 0 + } + + // 解析返回值 + vals, ok := result.([]interface{}) + if !ok || len(vals) != 2 { + return true, 0 + } + + allowed, _ := vals[0].(int64) + retryAfterStr, _ := vals[1].(string) + retryAfterSec, _ := strconv.ParseFloat(retryAfterStr, 64) + + if allowed == 1 { + return true, 0 + } + + retryAfter := time.Duration(retryAfterSec*1000) * time.Millisecond + return false, retryAfter +} + +// Stop 实现 Limiter 接口(Redis 不需要清理资源)。 +func (l *RedisLimiter) Stop() { + // Redis 客户端由外部管理,这里不需要操作 +} + +// getBucketConfig 根据 key 获取桶配置。 +func (l *RedisLimiter) getBucketConfig(key string) config.BucketConfig { + // 简化实现:默认使用 query 配置 + return l.config.Query +} + +// KeyPrefix 返回限流 key 的前缀。 +func KeyPrefix() string { + return "ratelimit:" +} + +// FormatKey 格式化限流 key。 +func FormatKey(userID, action string) string { + return fmt.Sprintf("%s%s:%s", KeyPrefix(), userID, action) +} + +// 编译期接口检查 +var _ Limiter = (*RedisLimiter)(nil) diff --git a/backend/internal/ratelimit/redis_bucket_test.go b/backend/internal/ratelimit/redis_bucket_test.go new file mode 100644 index 0000000..becc606 --- /dev/null +++ b/backend/internal/ratelimit/redis_bucket_test.go @@ -0,0 +1,228 @@ +package ratelimit + +import ( + "context" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/hhs/camtalk/internal/config" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// setupMiniRedis 创建一个内存 Redis 实例用于测试。 +func setupMiniRedis(t *testing.T) (*miniredis.Miniredis, *redis.Client) { + mr, err := miniredis.Run() + require.NoError(t, err) + + client := redis.NewClient(&redis.Options{ + Addr: mr.Addr(), + }) + + t.Cleanup(func() { + client.Close() + mr.Close() + }) + + return mr, client +} + +func TestRedisLimiter_Allow_FirstRequest(t *testing.T) { + _, client := setupMiniRedis(t) + + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 5, Rate: 0.2}, + } + limiter := NewRedisLimiter(client, cfg) + + ctx := context.Background() + allowed, retryAfter := limiter.Allow(ctx, "user1:query") + + assert.True(t, allowed) + assert.Equal(t, time.Duration(0), retryAfter) +} + +func TestRedisLimiter_Allow_ConsumeUntilEmpty(t *testing.T) { + _, client := setupMiniRedis(t) + + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 3, Rate: 0.2}, + } + limiter := NewRedisLimiter(client, cfg) + + ctx := context.Background() + key := "user1:query" + + // 连续消耗 3 个令牌 + for i := 0; i < 3; i++ { + allowed, _ := limiter.Allow(ctx, key) + assert.True(t, allowed, "request %d should be allowed", i+1) + } + + // 第 4 个请求应被拒绝 + allowed, retryAfter := limiter.Allow(ctx, key) + assert.False(t, allowed) + assert.Greater(t, retryAfter, time.Duration(0)) +} + +func TestRedisLimiter_Allow_DifferentKeys(t *testing.T) { + _, client := setupMiniRedis(t) + + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 2, Rate: 1.0}, + } + limiter := NewRedisLimiter(client, cfg) + + ctx := context.Background() + + // user1 消耗 2 个令牌 + allowed, _ := limiter.Allow(ctx, "user1:query") + assert.True(t, allowed) + allowed, _ = limiter.Allow(ctx, "user1:query") + assert.True(t, allowed) + + // user1 第 3 个被拒绝 + allowed, _ = limiter.Allow(ctx, "user1:query") + assert.False(t, allowed) + + // user2 应该不受影响 + allowed, _ = limiter.Allow(ctx, "user2:query") + assert.True(t, allowed) +} + +func TestRedisLimiter_Allow_RefillAfterWait(t *testing.T) { + _, client := setupMiniRedis(t) + + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 2, Rate: 10.0}, // 每秒 10 个令牌 + } + limiter := NewRedisLimiter(client, cfg) + + ctx := context.Background() + key := "user1:query" + + // 消耗 2 个令牌 + limiter.Allow(ctx, key) + limiter.Allow(ctx, key) + + // 真实等待 150ms(Lua 脚本使用系统时间) + time.Sleep(150 * time.Millisecond) + + // 应该补充了至少 1 个令牌 + allowed, _ := limiter.Allow(ctx, key) + assert.True(t, allowed) +} + +func TestRedisLimiter_Allow_CapacityLimit(t *testing.T) { + _, client := setupMiniRedis(t) + + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 3, Rate: 1.0}, + } + limiter := NewRedisLimiter(client, cfg) + + ctx := context.Background() + key := "user1:query" + + // 真实等待让桶"溢出" + time.Sleep(100 * time.Millisecond) + + // 但最多只能消耗 capacity 个令牌 + for i := 0; i < 3; i++ { + allowed, _ := limiter.Allow(ctx, key) + assert.True(t, allowed, "request %d should be allowed", i+1) + } + + // 第 4 个应被拒绝 + allowed, _ := limiter.Allow(ctx, key) + assert.False(t, allowed) +} + +func TestRedisLimiter_Allow_ZeroRate(t *testing.T) { + _, client := setupMiniRedis(t) + + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 1, Rate: 0.0}, + } + limiter := NewRedisLimiter(client, cfg) + + ctx := context.Background() + key := "user1:query" + + // 第一个通过 + allowed, _ := limiter.Allow(ctx, key) + assert.True(t, allowed) + + // 第二个被拒绝,retryAfter 应该很大 + allowed, retryAfter := limiter.Allow(ctx, key) + assert.False(t, allowed) + assert.Greater(t, retryAfter, 1*time.Hour) +} + +func TestRedisLimiter_Allow_KeyTTL(t *testing.T) { + mr, client := setupMiniRedis(t) + + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 5, Rate: 1.0}, + } + limiter := NewRedisLimiter(client, cfg) + + ctx := context.Background() + key := "user1:query" + + // 第一次请求 + limiter.Allow(ctx, key) + + // 验证 key 已设置 TTL + ttl := mr.TTL(key) + assert.Greater(t, ttl, time.Duration(0)) + assert.LessOrEqual(t, ttl, 600*time.Second) +} + +func TestRedisLimiter_Allow_FailOpen(t *testing.T) { + mr, client := setupMiniRedis(t) + + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 1, Rate: 1.0}, + } + limiter := NewRedisLimiter(client, cfg) + + ctx := context.Background() + + // 关闭 Redis 模拟故障 + mr.Close() + + // 应该 fail-open(允许请求) + allowed, retryAfter := limiter.Allow(ctx, "user1:query") + assert.True(t, allowed) + assert.Equal(t, time.Duration(0), retryAfter) +} + +func TestRedisLimiter_Stop(t *testing.T) { + _, client := setupMiniRedis(t) + + cfg := config.RateLimitConfig{ + Enabled: true, + Query: config.BucketConfig{Capacity: 1, Rate: 1.0}, + } + limiter := NewRedisLimiter(client, cfg) + + // Stop 应该不会 panic(即使多次调用) + limiter.Stop() + limiter.Stop() +} + +func TestFormatKey(t *testing.T) { + key := FormatKey("user123", "query") + assert.Equal(t, "ratelimit:user123:query", key) +} From ea00939c1326c0afcc0d1050d81da4a4c2ca76e2 Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sat, 20 Jun 2026 23:56:44 +0800 Subject: [PATCH 04/10] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=20Gin=20?= =?UTF-8?q?=E9=99=90=E6=B5=81=E4=B8=AD=E9=97=B4=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Middleware 函数返回 Gin 中间件 - keyFunc 参数支持灵活提取限流 key(IP/用户 ID 等) - 限流触发时返回 HTTP 429 + Retry-After header - 支持 nil limiter(跳过限流)和空 key(跳过限流) - 完整单元测试(6 个测试用例,全部通过) - 测试覆盖:允许、拒绝、nil limiter、空 key、keyFunc、Retry-After 舍入 --- backend/internal/ratelimit/middleware.go | 42 ++++ backend/internal/ratelimit/middleware_test.go | 196 ++++++++++++++++++ 2 files changed, 238 insertions(+) create mode 100644 backend/internal/ratelimit/middleware.go create mode 100644 backend/internal/ratelimit/middleware_test.go diff --git a/backend/internal/ratelimit/middleware.go b/backend/internal/ratelimit/middleware.go new file mode 100644 index 0000000..dceeee5 --- /dev/null +++ b/backend/internal/ratelimit/middleware.go @@ -0,0 +1,42 @@ +package ratelimit + +import ( + "fmt" + "net/http" + + "github.com/gin-gonic/gin" +) + +// Middleware 返回 Gin 中间件,按 key 维度限流。 +// keyFunc 从请求中提取限流 key(如 IP、用户 ID)。 +func Middleware(limiter Limiter, keyFunc func(*gin.Context) string) gin.HandlerFunc { + return func(c *gin.Context) { + if limiter == nil { + c.Next() + return + } + + key := keyFunc(c) + if key == "" { + // key 为空时跳过限流 + c.Next() + return + } + + allowed, retryAfter := limiter.Allow(c.Request.Context(), key) + + if !allowed { + // 设置 Retry-After header(秒) + c.Header("Retry-After", fmt.Sprintf("%d", int(retryAfter.Seconds()+0.5))) + + c.JSON(http.StatusTooManyRequests, gin.H{ + "code": "RATE_LIMITED", + "message": fmt.Sprintf("too many requests, retry after %s", retryAfter.Round(1)), + }) + c.Abort() + return + } + + c.Next() + } +} diff --git a/backend/internal/ratelimit/middleware_test.go b/backend/internal/ratelimit/middleware_test.go new file mode 100644 index 0000000..c9fd953 --- /dev/null +++ b/backend/internal/ratelimit/middleware_test.go @@ -0,0 +1,196 @@ +package ratelimit + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// mockLimiter 用于测试的 mock 限流器。 +type mockLimiter struct { + allowFunc func(ctx context.Context, key string) (bool, time.Duration) +} + +func (m *mockLimiter) Allow(ctx context.Context, key string) (bool, time.Duration) { + if m.allowFunc != nil { + return m.allowFunc(ctx, key) + } + return true, 0 +} + +func (m *mockLimiter) Stop() {} + +// 编译期接口检查 +var _ Limiter = (*mockLimiter)(nil) + +func TestMiddleware_Allow(t *testing.T) { + gin.SetMode(gin.TestMode) + + limiter := &mockLimiter{ + allowFunc: func(ctx context.Context, key string) (bool, time.Duration) { + return true, 0 + }, + } + + router := gin.New() + router.Use(Middleware(limiter, func(c *gin.Context) string { + return "user1:test" + })) + router.GET("/test", func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + }) + + req := httptest.NewRequest(http.MethodGet, "/test", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + + var resp map[string]interface{} + err := json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + assert.Equal(t, "ok", resp["status"]) +} + +func TestMiddleware_Deny(t *testing.T) { + gin.SetMode(gin.TestMode) + + limiter := &mockLimiter{ + allowFunc: func(ctx context.Context, key string) (bool, time.Duration) { + return false, 5 * time.Second + }, + } + + router := gin.New() + router.Use(Middleware(limiter, func(c *gin.Context) string { + return "user1:test" + })) + router.GET("/test", func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + }) + + req := httptest.NewRequest(http.MethodGet, "/test", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + // 验证返回 429 + assert.Equal(t, http.StatusTooManyRequests, w.Code) + + // 验证 Retry-After header + assert.Equal(t, "5", w.Header().Get("Retry-After")) + + // 验证响应体 + var resp map[string]interface{} + err := json.Unmarshal(w.Body.Bytes(), &resp) + require.NoError(t, err) + assert.Equal(t, "RATE_LIMITED", resp["code"]) + assert.Contains(t, resp["message"], "retry after") +} + +func TestMiddleware_NilLimiter(t *testing.T) { + gin.SetMode(gin.TestMode) + + router := gin.New() + router.Use(Middleware(nil, func(c *gin.Context) string { + return "user1:test" + })) + router.GET("/test", func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + }) + + req := httptest.NewRequest(http.MethodGet, "/test", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + // nil limiter 应该放行 + assert.Equal(t, http.StatusOK, w.Code) +} + +func TestMiddleware_EmptyKey(t *testing.T) { + gin.SetMode(gin.TestMode) + + limiter := &mockLimiter{ + allowFunc: func(ctx context.Context, key string) (bool, time.Duration) { + // 不应该被调用 + t.Error("Allow should not be called with empty key") + return false, 0 + }, + } + + router := gin.New() + router.Use(Middleware(limiter, func(c *gin.Context) string { + return "" // 返回空 key + })) + router.GET("/test", func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + }) + + req := httptest.NewRequest(http.MethodGet, "/test", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + // 空 key 应该放行 + assert.Equal(t, http.StatusOK, w.Code) +} + +func TestMiddleware_KeyFunc(t *testing.T) { + gin.SetMode(gin.TestMode) + + var capturedKey string + limiter := &mockLimiter{ + allowFunc: func(ctx context.Context, key string) (bool, time.Duration) { + capturedKey = key + return true, 0 + }, + } + + router := gin.New() + router.Use(Middleware(limiter, func(c *gin.Context) string { + // 从 query 参数提取 user_id + userID := c.Query("user_id") + return userID + ":test" + })) + router.GET("/test", func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + }) + + req := httptest.NewRequest(http.MethodGet, "/test?user_id=user123", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "user123:test", capturedKey) +} + +func TestMiddleware_RetryAfterRounding(t *testing.T) { + gin.SetMode(gin.TestMode) + + limiter := &mockLimiter{ + allowFunc: func(ctx context.Context, key string) (bool, time.Duration) { + return false, 2500 * time.Millisecond // 2.5 秒 + }, + } + + router := gin.New() + router.Use(Middleware(limiter, func(c *gin.Context) string { + return "user1:test" + })) + router.GET("/test", func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"status": "ok"}) + }) + + req := httptest.NewRequest(http.MethodGet, "/test", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + assert.Equal(t, http.StatusTooManyRequests, w.Code) + // 2.5 秒向上取整为 3 秒 + assert.Equal(t, "3", w.Header().Get("Retry-After")) +} From 7adf81c6e5960ceb383d8cc025e06ddad00a1b41 Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sun, 21 Jun 2026 00:00:24 +0800 Subject: [PATCH 05/10] =?UTF-8?q?feat:=20=E9=9B=86=E6=88=90=E9=99=90?= =?UTF-8?q?=E6=B5=81=E5=99=A8=E5=88=B0=E6=9C=8D=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - main.go 初始化限流器(根据 Redis 可用性选择内存/Redis 实现) - WebSocket handler 添加 query 消息限流(按 userID) - Auth API 添加登录/注册限流(按 IP) - refresh 和 logout 不限流(避免影响正常用户操作) - 修复所有测试(传递 nil limiter 参数) - 所有测试通过(包括 ws 和 api 集成测试) --- CLAUDE.md | 210 ++++++++++++++++++++++++---- backend/cmd/server/main.go | 22 ++- backend/internal/api/auth.go | 22 ++- backend/internal/api/auth_test.go | 4 +- backend/internal/ws/handler.go | 20 ++- backend/internal/ws/handler_test.go | 6 +- 6 files changed, 240 insertions(+), 44 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index f0b3da7..6d11099 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -6,7 +6,25 @@ CamTalk 是一款多模态实时 AI 视觉对话助手。用户通过摄像头和麦克风与 AI 交互,AI 理解视觉场景和语音输入后,以文字和语音形式给出自然回应。 -> **文档优先原则:** 执行任何开发任务前,先读取 `docs/` 下的相关设计文档(架构、接口、技术选型等),以文档为最高依据。代码实现应与文档一致;若有偏差,优先更新文档(尤其是接口文档)。 +> **文档优先原则:** 执行任何开发任务前,先读取 `docs/` 下的相关设计文档(架构、接口、技术选型等),以文档为最高依据。代码实现应与文档一致;若有偏差,优先更新文档(尤其是接口文档)。`docs/Eino/` 下有完整的 Eino 框架文档(~75 个 markdown 文件),可作为参考。 + +**核心设计文档:** + +| 文档 | 内容 | +|------|------| +| `docs/01-架构设计.md` | 三层架构、技术栈、数据库设计、部署方案 | +| `docs/02-接口文档.md` | WebSocket 协议、REST API、AI 服务层、编排器、配置管理 | +| `docs/03-技术选型.md` | AI 服务栈、持久化层、前端边缘处理选型 | +| `docs/04-用户故事.md` | 用户场景与优先级 | +| `docs/05-语音交互.md` | VAD → STT → LLM → TTS 全链路 | +| `docs/06-视觉理解.md` | 帧采样、关键帧检测、多模态输入 | +| `docs/07-成本控制.md` | 采样策略、端云协同、模型分级 | +| `docs/08-功能创意.md` | 功能创意与规划 | +| `docs/09-技术名词解释.md` | 术语定义(VAD/STT/TTS/Token/JWT 等) | +| `docs/10-Eino重构方案.md` | Eino Graph 迁移方案与决策记录 | +| `docs/11-Eino框架技术文档.md` | Eino 框架使用指南 | +| `docs/12-鉴权体系设计.md` | JWT 双 token 轮转详细设计 | +| `docs/13-令牌桶限流设计.md` | 令牌桶限流详细设计(需同步实现) | ## 架构 @@ -16,30 +34,65 @@ CamTalk 是一款多模态实时 AI 视觉对话助手。用户通过摄像头 2. **Go 网关**(Gin, gorilla/websocket, Viper, Zap)—— WebSocket 服务器、会话管理、AI 编排(基于 CloudWeGo Eino Graph)。每个 WebSocket 连接一个 goroutine。 3. **云端 AI 服务** —— 通过 OpenAI 兼容接口可灵活切换。默认:DashScope qwen3-vl-plus(LLM)、MiMo ASR(STT)、MiMo TTS(TTS)。仅通过 Go 网关访问,浏览器不直连。 -**关键模式**:AI 编排基于 Eino Graph 声明式 DAG(`START → STT → History → ChatModel → Msg2Str → Splitter → TTS → Done → END`),LLM token 通过 Callback 实时推送,TTS 逐句合成并行推送,最小化感知延迟。 +**关键模式**:AI 编排基于 Eino Graph 声明式 DAG(6 节点线性流水线:`START → STT → History → ChatModel → Msg2Str → Splitter → TTS → Done → END`),LLM token 通过 Callback 实时推送,TTS 逐句合成并行推送,最小化感知延迟。 -**存储**:三级存储架构(TieredManager)—— L1 Memory → L2 Redis → L3 PostgreSQL,自动降级。Repository 接口模式(UserRepository、MessageRepository、SessionRepository),PostgreSQL + 内存双实现。 +### Eino Graph 节点详解 + +| 节点 | 类型 | 文件 | 职责 | +|------|------|------|------| +| STT | `InvokableLambda` | `backend/internal/eino/nodes_stt.go` | 语音识别或文本直通(text-only 跳过 STT) | +| History | `InvokableLambda` | `backend/internal/eino/nodes_history.go` | 构建 System Prompt + 对话历史 + 用户输入 + 图像 | +| ChatModel | ChatModel 节点 | `backend/internal/eino/graph.go` | 调用 DashScope qwen3-vl-plus(OpenAI 兼容协议) | +| Msg2Str | `TransformableLambda` | `backend/internal/eino/nodes_splitter.go` | 将 ChatModel 流式 Message 转为字符串流 | +| Splitter | `TransformableLambda` | `backend/internal/eino/nodes_splitter.go` | 按句子分隔符(`。!?\n.!?`)拆分文本流 | +| TTS | `TransformableLambda` | `backend/internal/eino/nodes_tts.go` | 逐句合成语音并推送 `tts_audio` | +| Done | `InvokableLambda` | `backend/internal/eino/nodes_done.go` | 发送 `llm_done`、收集最终输出 | + +**跨节点状态**:`PipelineState`(`backend/internal/eino/state.go`),通过 `context.WithValue` 在节点间传递 FullResponse、TranscribedText、TokenUsage、SessionID、RequestID。 + +**Callback**:`BuildCallbackHandler`(`backend/internal/eino/callback.go`)挂载到 ChatModel 的 `OnEndWithStreamOutput`,每收到一个 LLM token 立即通过 `sender.SendLLMChunk()` 推送到客户端。 + +**适配器**:`EinoOrchestrator`(`backend/internal/eino/adapter.go`)包装 Graph,实现 `orchestrator.Orchestrator` 接口,负责解码 Base64 图像/音频、构建输入、注入上下文、运行流式推理、持久化消息。 + +### 会话存储(TieredManager) + +三级存储:**L1 Memory → L2 Redis → L3 PostgreSQL**(`backend/internal/session/tiered.go`) + +- **读路径**:L1 命中直接返回;未命中尝试 L2 Redis → 回填 L1;L3 通过 L1 的 `FindByID` 降级读取 +- **写路径**:L1 同步写入 → L2 同步写(失败 soft-warn)→ L3 异步 goroutine 写(使用 `context.Background()` 防止请求取消丢失) +- **降级**:后台协程每 30 秒 ping Redis,Redis 不可用时自动跳过 L2 操作;恢复后自动重新启用 +- **TTL**:Session 默认 30 分钟,MaxHistory 20 条;L1 后台协程每分钟清理过期 session + +Repository 接口模式:`UserRepository`、`MessageRepository`、`SessionRepository`,均有 PostgreSQL 和内存双实现。 + +### 鉴权 + +JWT 双 token 轮转认证(HMAC-SHA256): +- Access Token:默认 120 分钟 TTL,Bearer header 传递 +- Refresh Token:默认 7 天 TTL,带 jti(UUID),Hash 存储在 Redis/PostgreSQL +- 轮转:Refresh 时旧 token hash 删除,新 pair 生成;若 JWT 有效但 DB hash 缺失 → 判定为重放攻击 → 吊销该用户所有 refresh token +- `CachedUserRepository`(`backend/internal/store/cached_user.go`):装饰器模式,Redis 缓存 refresh token hash,Read-Through / Write-Through,Redis 故障软降级 ## 技术栈 | 层级 | 技术 | |------|------| -| 前端 | React 18, TypeScript, Vite, @ricky0123/vad-web | -| 后端 | Go, Gin, gorilla/websocket, Viper, Zap | +| 前端 | React 18, TypeScript, Vite, @ricky0123/vad-web, onnxruntime-web | +| 后端 | Go 1.25+ (go.mod 最低要求; Dockerfile 构建用 golang:1.26-alpine), Gin, gorilla/websocket, Viper, Zap | | AI 编排 | CloudWeGo Eino Graph(声明式 DAG 编排) | | LLM | DashScope qwen3-vl-plus(默认,通过 eino-ext OpenAI ChatModel 接入) | | STT | MiMo ASR(默认) / Deepgram | | TTS | MiMo TTS(默认) / OpenAI TTS | +| 存储 | PostgreSQL 15 + Redis 7(通过 TieredManager 三级存储) | ## 构建与运行命令 ```bash # 前端 cd frontend && npm install -npm run dev # Vite 开发服务器 -npm run build # 生产构建 -npm run lint # ESLint 检查 -npm run test # Vitest 测试 +npm run dev # Vite 开发服务器(含 /ws、/api 代理到 localhost:8080) +npm run build # 生产构建(tsc -b && vite build) +npm run lint # ESLint 检查(flat config, TypeScript strict) # 后端 cd backend && go mod download @@ -48,9 +101,75 @@ go build -o bin/camtalk ./cmd/server go test ./... # 运行所有测试 go test -run TestName ./path # 运行单个测试 go vet ./... # 静态分析 + +# Docker 部署(生产环境) +./deploy.sh build # 构建所有镜像 +./deploy.sh up # 启动 4 个服务 +./deploy.sh restart # down + up +./deploy.sh logs [service] # 查看日志 +./deploy.sh status # 查看服务状态 ``` -基础设施:三级存储架构(L1 Memory → L2 Redis → L3 PostgreSQL),通过配置控制启用层级。 +> **注意**:前端目前没有测试基础设施(无 vitest 配置、无测试文件)。后端使用 `testing` + `testify`(assert/require/mock)测试,编译期接口检查 `var _ Interface = (*Impl)(nil)`。 + +## 配置系统 + +配置文件:`backend/config.yaml`(默认值),可被 `config.{env}.yaml` 覆盖。 + +**优先级(从低到高)**:默认值 → `config.yaml` → `config.{env}.yaml`(由 `APP_ENV` 环境变量决定加载哪个 env 特定文件)→ `.env` 文件 → 环境变量 + +**主要 `CAMTALK_` 环境变量**(模板见 `backend/.env.example`): + +| 变量 | 用途 | +|------|------| +| `APP_ENV` | 运行环境(dev/prod),决定加载 `config.{env}.yaml` | +| `CAMTALK_AI_STT_API_KEY` | STT API Key | +| `CAMTALK_AI_LLM_API_KEY` | LLM API Key | +| `CAMTALK_AI_TTS_API_KEY` | TTS API Key | +| `CAMTALK_AUTH_JWT_SECRET` | JWT 签名密钥 | +| `CAMTALK_STORAGE_DSN` | PostgreSQL 连接串 | +| `CAMTALK_STORAGE_REDIS_ENABLED` | 启用 Redis(true/false) | +| `CAMTALK_STORAGE_PERSISTENCE_ENABLED` | 启用 PostgreSQL(true/false) | +| `CAMTALK_REDIS_ADDR` | Redis 地址 | +| `CAMTALK_REDIS_PASSWORD` | Redis 密码 | + +**最小启动**(至少需要一个 AI 服务的 API Key): +```bash +CAMTALK_AI_LLM_API_KEY=sk-xxx CAMTALK_AI_STT_API_KEY=xxx go run ./cmd/server +``` + +## Docker 部署 + +`docker-compose.yml` 定义 4 个服务(`camtalk-net` 桥接网络): + +| 服务 | 镜像/构建 | 端口 | 说明 | +|------|----------|------|------| +| `frontend` | 构建 `./frontend/Dockerfile`(node:22-alpine → nginx:stable-alpine) | 9000:80 | React SPA,反向代理 /api 和 /ws 到 backend | +| `backend` | 构建 `./backend/Dockerfile`(golang:1.26-alpine → alpine:3.20) | 内部 8080 | Go 网关,静态链接二进制 `-ldflags="-s -w"` | +| `postgres` | `postgres:15-alpine` | 内部 5432 | 数据库 `camtalk`,挂载 `./backend/migrations/` 到 initdb | +| `redis` | `redis:7-alpine` | 内部 6379 | 会话缓存,AOF 持久化 | + +后端容器依赖 postgres + redis 健康检查通过后启动。所有服务 `restart: unless-stopped`。密钥通过 `--env-file /opt/camtalk/.env` 注入。 + +**前端 nginx 特殊配置**:设置 `Cross-Origin-Opener-Policy` 和 `Cross-Origin-Embedder-Policy` 头(`SharedArrayBuffer` 需要,ONNX WASM 推理依赖)。 + +## CI/CD + +使用 **Gitea Actions**(`.gitea/workflows/deploy.yml`),自托管 runner(标签 `aliyun`)。 + +触发条件:push 到 `main` 或 `v2` 分支。流程:rsync 代码到 `/root/camtalk`,执行 `deploy.sh build` → `deploy.sh restart`。 + +## 数据库迁移 + +嵌入式 SQL 迁移系统(`backend/internal/store/migrate.go`),SQL 文件在 `backend/migrations/`: + +| 迁移 | 内容 | +|------|------| +| `001_users` | `users` 表(UUID PK)+ `refresh_tokens` 表(FK → users) | +| `002_messages` | `messages` 表(BIGSERIAL PK, session_id UUID, 游标分页索引) | +| `003_sessions` | `sessions` 表(UUID PK, user_id UUID, config JSONB, 时间排序索引) | + +迁移文件通过 Go 1.16+ `//go:embed` 嵌入二进制,启动时自动执行。通过 `schema_migrations` 表追踪版本,已应用的迁移跳过。同时挂载到 PostgreSQL 容器的 `/docker-entrypoint-initdb.d` 作为备用初始化路径。 ## WebSocket 协议 @@ -58,12 +177,16 @@ go vet ./... # 静态分析 所有消息为 JSON 文本帧,统一信封格式 `{type, request_id?, timestamp?}`。完整契约见 `docs/02-接口文档.md`。 +**认证**:WebSocket 连接通过 query param `token`(Access Token)认证,不走 HTTP `Authorization` header。服务端在升级时校验 JWT,失败返回 401。 + **客户端 → 服务端**:`query`(图像 Base64 + 音频 Base64)、`config`、`interrupt`、`ping` **服务端 → 客户端**:`connected`、`stt_result`、`llm_chunk`、`llm_done`、`tts_audio`、`error`、`pong` **心跳**:客户端每 30 秒 ping,服务端 60 秒无 ping 断开连接。 **重连**:指数退避 + 抖动 —— 1s, 2s, 4s, 8s… 最大 30s。 +**前端 WebSocket 实现**:`CamTalkWebSocket` 单例类(`frontend/src/lib/websocket.ts`),基于订阅模式(`onMessage`/`onStatusChange` 返回取消订阅函数),自动处理心跳和重连。 + ## REST API(辅助) - `GET /api/health` — 健康检查(版本、运行时间、活跃会话数) @@ -74,7 +197,8 @@ go vet ./... # 静态分析 - `GET /api/conversations` — 对话列表 - `POST /api/conversations` — 创建对话 - `GET/PATCH/DELETE /api/conversations/:id` — 对话详情/改标题/删除 -- `GET /api/conversations/:id/messages` — 获取对话消息 +- `GET /api/conversations/:id/messages` — 获取对话消息(游标分页) +- `POST/DELETE /api/sessions` — 会话管理 ## 错误码 @@ -86,39 +210,63 @@ go vet ./... # 静态分析 |------|------| | `LandingPage` | 未登录时的着陆页,内嵌 LoginModal 登录/注册弹窗 | | `AuthPage` | 登录/注册表单(备用) | -| `CameraManager` | 摄像头流采集 | -| `MicManager` | 麦克风音频采集 | -| `EdgeProcessor` | VAD + 关键帧检测(Canvas 像素比较) | -| `WebSocketManager` | WebSocket 连接生命周期管理 | -| `ChatPanel` | 消息展示、流式回复、文本输入、场景选择 | -| `VideoPreview` | 摄像头画面预览 | +| `CameraManager` | 摄像头流采集(`useCamera` hook:640x480, facingMode: environment) | +| `MicManager` | 麦克风音频采集(`useMicrophone` hook:16kHz 单声道) | +| `EdgeProcessor` | VAD(`useVAD` hook:@ricky0123/vad-web)+ 关键帧检测(Canvas 像素比较,160x120 降采样) | +| `WebSocketManager` | WebSocket 连接生命周期管理(桥接 `wsClient` 单例到 React 状态) | +| `ChatPanel` | 消息展示、流式回复、文本输入、场景选择(5 种场景卡片) | +| `VideoPreview` | 摄像头画面预览(forwardRef `