feat: 添加 Redis Docker 容器编排及三级存储架构
- docker-compose.yml 新增 Redis 服务及 camtalk-net 桥接网络 - 实现 TieredManager 三级存储(L1 内存 → L2 Redis → L3 PostgreSQL) - config.go 新增 Redis/Persistence 配置类型及环境变量绑定 - 修复 CAMTALK_REDIS_ADDR 环境变量未被 Viper 绑定的问题 - .env.example 更新为三级存储配置并标注开发/部署地址差异
This commit is contained in:
@@ -10,6 +10,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"github.com/hhs/camtalk/internal/api"
|
||||
"github.com/hhs/camtalk/internal/auth"
|
||||
@@ -47,7 +48,7 @@ func main() {
|
||||
"addr", cfg.Server.Addr(),
|
||||
)
|
||||
|
||||
// 初始化存储层(条件初始化 PostgreSQL)
|
||||
// 初始化存储层(三级存储架构:L1 内存 → L2 Redis → L3 PostgreSQL)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
@@ -55,12 +56,17 @@ func main() {
|
||||
var msgRepo store.MessageRepository
|
||||
var sessRepo store.SessionRepository
|
||||
|
||||
if cfg.Storage.Driver == "postgres" {
|
||||
if cfg.Storage.DSN == "" {
|
||||
logger.Log.Fatalw("storage.dsn is required when storage.driver is postgres",
|
||||
// L3: PostgreSQL(冷数据持久化层)
|
||||
dsn := cfg.Storage.Persistence.DSN
|
||||
if dsn == "" {
|
||||
dsn = cfg.Storage.DSN // 兼容旧配置
|
||||
}
|
||||
if cfg.Storage.Persistence.Enabled && cfg.Storage.Persistence.Driver == "postgres" {
|
||||
if dsn == "" {
|
||||
logger.Log.Fatalw("storage.persistence.dsn is required when persistence is enabled",
|
||||
"hint", "set CAMTALK_STORAGE_DSN environment variable")
|
||||
}
|
||||
pool, err := store.NewPostgresPool(ctx, cfg.Storage.DSN)
|
||||
pool, err := store.NewPostgresPool(ctx, dsn)
|
||||
if err != nil {
|
||||
logger.Log.Fatalw("failed to connect to postgres", "error", err)
|
||||
}
|
||||
@@ -74,27 +80,72 @@ func main() {
|
||||
userRepo = store.NewPgUserRepository(pool)
|
||||
msgRepo = store.NewPgMessageRepository(pool)
|
||||
sessRepo = store.NewPgSessionRepository(pool)
|
||||
logger.Log.Infow("postgres storage initialized", "driver", cfg.Storage.Driver)
|
||||
logger.Log.Infow("L3 PostgreSQL storage initialized", "driver", cfg.Storage.Persistence.Driver)
|
||||
} else {
|
||||
userRepo = store.NewMemUserRepository()
|
||||
logger.Log.Info("using in-memory storage")
|
||||
logger.Log.Info("using in-memory user storage")
|
||||
}
|
||||
|
||||
// 初始化 Session Manager
|
||||
// L2: Redis(热数据分布式会话层)
|
||||
var redisMgr *session.RedisManager
|
||||
if cfg.Storage.Redis.Enabled {
|
||||
rdb := redis.NewClient(&redis.Options{
|
||||
Addr: cfg.Redis.Addr,
|
||||
Password: cfg.Redis.Password,
|
||||
DB: cfg.Redis.DB,
|
||||
})
|
||||
// 验证 Redis 连接
|
||||
if err := rdb.Ping(ctx).Err(); err != nil {
|
||||
logger.Log.Fatalw("failed to connect to redis", "error", err)
|
||||
}
|
||||
redisMgr = session.NewRedisManager(
|
||||
rdb,
|
||||
time.Duration(cfg.Session.TTL)*time.Minute,
|
||||
cfg.Session.MaxHistory,
|
||||
)
|
||||
logger.Log.Infow("L2 Redis storage initialized",
|
||||
"addr", cfg.Redis.Addr,
|
||||
"db", cfg.Redis.DB)
|
||||
}
|
||||
|
||||
// 初始化 Session Manager(三级存储)
|
||||
var sessionMgr session.Manager
|
||||
var sessionOpts []session.Option
|
||||
if msgRepo != nil {
|
||||
sessionOpts = append(sessionOpts, session.WithMessageRepository(msgRepo))
|
||||
if cfg.Storage.Redis.Enabled {
|
||||
// L1 + L2 + L3 三级存储
|
||||
var tieredOpts []session.TieredOption
|
||||
if sessRepo != nil {
|
||||
tieredOpts = append(tieredOpts, session.WithTieredSessionRepository(sessRepo))
|
||||
}
|
||||
if msgRepo != nil {
|
||||
tieredOpts = append(tieredOpts, session.WithTieredMessageRepository(msgRepo))
|
||||
}
|
||||
tieredMgr := session.NewTieredManager(
|
||||
time.Duration(cfg.Session.TTL)*time.Minute,
|
||||
cfg.Session.MaxHistory,
|
||||
redisMgr,
|
||||
tieredOpts...,
|
||||
)
|
||||
sessionMgr = tieredMgr
|
||||
defer tieredMgr.Stop()
|
||||
logger.Log.Info("session manager initialized with L1+L2+L3 tiered storage")
|
||||
} else {
|
||||
// L1 + L3 两级存储(无 Redis)
|
||||
var sessionOpts []session.Option
|
||||
if msgRepo != nil {
|
||||
sessionOpts = append(sessionOpts, session.WithMessageRepository(msgRepo))
|
||||
}
|
||||
if sessRepo != nil {
|
||||
sessionOpts = append(sessionOpts, session.WithSessionRepository(sessRepo))
|
||||
}
|
||||
memMgr := session.NewMemoryManager(
|
||||
time.Duration(cfg.Session.TTL)*time.Minute,
|
||||
cfg.Session.MaxHistory,
|
||||
sessionOpts...,
|
||||
)
|
||||
sessionMgr = memMgr
|
||||
defer memMgr.Stop()
|
||||
logger.Log.Info("session manager initialized with L1+L3 storage (Redis disabled)")
|
||||
}
|
||||
if sessRepo != nil {
|
||||
sessionOpts = append(sessionOpts, session.WithSessionRepository(sessRepo))
|
||||
}
|
||||
sessionMgr = session.NewMemoryManager(
|
||||
time.Duration(cfg.Session.TTL)*time.Minute,
|
||||
cfg.Session.MaxHistory,
|
||||
sessionOpts...,
|
||||
)
|
||||
defer sessionMgr.(*session.MemoryManager).Stop()
|
||||
|
||||
// 初始化 AI 服务
|
||||
logger.Log.Infow("initializing AI services",
|
||||
|
||||
Reference in New Issue
Block a user