lan
d9aa4b46c3
- Introduce interfaces for all major services (JWT, PostAI, Comment, Message, Notification, QRCodeLogin, Upload, Vote, etc.) to support dependency inversion. - Move query parameters from `internal/dto` to a new `internal/query` package to separate request payloads from data transfer objects. - Refactor `internal/router` to use embedded `RouterDeps` for cleaner dependency management. - Decouple handlers from repositories by injecting services instead of direct repository access, ensuring proper layering. - Improve database initialization by moving it from `internal/model` to `internal/database`. - Optimize message decryption by implementing a more efficient `BatchDecrypt` method in `MessageEncryptor` using a worker pool. - Enhance error handling and security by implementing fail-fast checks for encryption key length during startup. - Clean up unused code, including the `avatar` package and several unused DTOs.
106 lines
3.3 KiB
Go
106 lines
3.3 KiB
Go
package cache
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
|
||
redisPkg "with_you/internal/pkg/redis"
|
||
)
|
||
|
||
const seqKeyTTL = 24 * time.Hour
|
||
|
||
// SeqBufferConfig seq 预分配配置（保留结构以兼容配置文件，实际不再使用缓冲）
|
||
type SeqBufferConfig struct {
|
||
Enabled bool `mapstructure:"enabled"`
|
||
PrivateBufferSize int `mapstructure:"private_buffer_size"`
|
||
GroupBufferSize int `mapstructure:"group_buffer_size"`
|
||
LockTimeoutMs int `mapstructure:"lock_timeout_ms"`
|
||
MaxRetries int `mapstructure:"max_retries"`
|
||
}
|
||
|
||
// SeqBufferManager seq 分配管理器（简化版：每条消息一个 Redis INCR，无本地缓冲）
|
||
type SeqBufferManager struct {
|
||
rdb *redis.Client
|
||
config SeqBufferConfig
|
||
repo ConversationRepository
|
||
msgRepo MessageRepository
|
||
}
|
||
|
||
// NewSeqBufferManager 创建 seq 分配管理器
|
||
func NewSeqBufferManager(rdb *redis.Client, cfg SeqBufferConfig, repo ConversationRepository, msgRepo MessageRepository) *SeqBufferManager {
|
||
return &SeqBufferManager{
|
||
rdb: rdb,
|
||
config: cfg,
|
||
repo: repo,
|
||
msgRepo: msgRepo,
|
||
}
|
||
}
|
||
|
||
// GetNextSeq 获取下一个 seq（单次 Redis INCR，原子且无竞态）
|
||
func (m *SeqBufferManager) GetNextSeq(ctx context.Context, convID string, _ bool) (int64, error) {
|
||
seqKey := MessageSeqKey(convID)
|
||
|
||
// 1. 尝试 INCR（key 存在时直接递增，原子操作）
|
||
result, err := nextSeqLua.Run(ctx, m.rdb, []string{seqKey}).Int64()
|
||
if err == nil {
|
||
if result > 0 {
|
||
m.rdb.Expire(ctx, seqKey, seqKeyTTL)
|
||
return result, nil
|
||
}
|
||
// result == 0: 冷启动，需从 DB 初始化
|
||
return m.initAndIncr(ctx, convID, seqKey)
|
||
}
|
||
|
||
// Redis 不可用，降级到 DB
|
||
zap.L().Warn("seq INCR failed, falling back to DB", zap.String("convID", convID), zap.Error(err))
|
||
if m.msgRepo != nil {
|
||
if dbSeq, dbErr := m.msgRepo.GetNextSeq(convID); dbErr == nil {
|
||
return dbSeq, nil
|
||
}
|
||
}
|
||
return 0, fmt.Errorf("seq allocation failed for %s: %w", convID, err)
|
||
}
|
||
|
||
// initAndIncr 冷启动：从 DB 加载 last_seq，然后用 SETNX+INCR 原子初始化
|
||
func (m *SeqBufferManager) initAndIncr(ctx context.Context, convID, seqKey string) (int64, error) {
|
||
if m.repo == nil {
|
||
return 0, fmt.Errorf("conversation repository not configured for seq init")
|
||
}
|
||
|
||
conv, err := m.repo.GetConversationByID(convID)
|
||
if err != nil {
|
||
return 0, fmt.Errorf("load conversation for seq init: %w", err)
|
||
}
|
||
|
||
initVal := conv.LastSeq
|
||
result, err := initSeqLua.Run(ctx, m.rdb, []string{seqKey}, initVal).Int64()
|
||
if err != nil {
|
||
return 0, fmt.Errorf("initSeqLua failed: %w", err)
|
||
}
|
||
|
||
m.rdb.Expire(ctx, seqKey, seqKeyTTL)
|
||
return result, nil
|
||
}
|
||
|
||
// InvalidateBuffer no-op（保留接口兼容）
|
||
func (m *SeqBufferManager) InvalidateBuffer(convID string) {}
|
||
|
||
// IsEnabled 是否启用
|
||
func (m *SeqBufferManager) IsEnabled() bool {
|
||
return m.config.Enabled && m.rdb != nil
|
||
}
|
||
|
||
// NewSeqBufferManagerFromRedisPkg 从 pkg/redis.Client 创建 SeqBufferManager
|
||
func NewSeqBufferManagerFromRedisPkg(redisPkgClient *redisPkg.Client, cfg SeqBufferConfig, repo ConversationRepository, msgRepo MessageRepository) *SeqBufferManager {
|
||
if redisPkgClient == nil {
|
||
zap.L().Warn("SeqBuffer: redis client is nil, seq pre-allocation disabled")
|
||
return &SeqBufferManager{config: cfg, repo: repo, msgRepo: msgRepo}
|
||
}
|
||
rdb := redisPkgClient.GetClient()
|
||
return NewSeqBufferManager(rdb, cfg, repo, msgRepo)
|
||
}
|