Files
backend/internal/wire/infrastructure.go
lan c630cbf4d0
All checks were successful
Build Backend / build (push) Successful in 2m0s
Build Backend / build-docker (push) Successful in 1m26s
feat(ws): implement WebSocket cluster mode with Redis Pub/Sub
Introduce a new WebSocket messaging architecture that supports both standalone and cluster modes. This allows for horizontal scaling of WebSocket servers by using Redis Pub/Sub to synchronize messages across multiple instances.

Key changes:
- Added `ws.MessagePublisher` interface to abstract message distribution.
- Implemented `ws.Bus` to handle cluster-mode messaging via Redis.
- Added `ws.OnlineTracker` to manage user online status across the cluster.
- Refactored multiple services (Chat, Group, Push, Call, etc.) to use the new `MessagePublisher` instead of a concrete `ws.Hub`.
- Added WebSocket configuration options (mode, instance ID, channel, TTL, heartbeat) to `config.yaml` and `config.go`.
- Updated dependency injection with Wire to support the new publisher and Redis client.
- Improved logging by replacing standard `log` with `zap` in several service components.
2026-05-06 12:39:11 +08:00

278 lines
7.7 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package wire
import (
"time"
"with_you/internal/cache"
"with_you/internal/config"
"with_you/internal/model"
"with_you/internal/pkg/crypto"
"with_you/internal/pkg/email"
"with_you/internal/pkg/hook"
"with_you/internal/pkg/jpush"
"with_you/internal/pkg/openai"
"with_you/internal/pkg/redis"
"with_you/internal/pkg/s3"
"with_you/internal/pkg/tencent"
"with_you/internal/pkg/ws"
"with_you/internal/repository"
"with_you/internal/service"
"github.com/google/wire"
"go.uber.org/zap"
"gorm.io/gorm"
)
// InfrastructureSet 基础设施 Provider Set
var InfrastructureSet = wire.NewSet(
// 配置
ProvideConfig,
// 日志
ProvideLogger,
// 数据库
ProvideDB,
// 事务管理器
ProvideTransactionManager,
// Redis 和缓存
ProvideRedisClient,
ProvideCache,
// 钩子系统
ProvideHookManager,
ProvideBuiltinHooks,
ProvideModerationHooks,
// WebSocket MessagePublisher
ProvideWSMessagePublisher,
// 外部服务客户端
ProvideOpenAIClient,
ProvideTencentClient,
ProvideEmailClient,
ProvideS3Client,
ProvideJPushClient,
// 消息加密器
ProvideMessageEncryptor,
)
// ProvideConfig 加载配置
func ProvideConfig() (*config.Config, error) {
return config.Load("configs/config.yaml")
}
// ProvideDB 提供数据库连接
func ProvideDB(cfg *config.Config) (*gorm.DB, error) {
return model.NewDB(&cfg.Database)
}
// ProvideRedisClient 提供 Redis 客户端
func ProvideRedisClient(cfg *config.Config) *redis.Client {
client, err := redis.New(&cfg.Redis)
if err != nil {
zap.L().Warn("Failed to connect to Redis, falling back to memory cache",
zap.Error(err),
)
return nil
}
return client
}
// ProvideCache 提供缓存实例
func ProvideCache(cfg *config.Config, redisClient *redis.Client) cache.Cache {
cache.Configure(cache.Settings{
Enabled: cfg.Cache.Enabled,
KeyPrefix: cfg.Cache.KeyPrefix,
DefaultTTL: time.Duration(cfg.Cache.DefaultTTL) * time.Second,
NullTTL: time.Duration(cfg.Cache.NullTTL) * time.Second,
JitterRatio: cfg.Cache.JitterRatio,
PostListTTL: time.Duration(cfg.Cache.Modules.PostList) * time.Second,
ConversationTTL: time.Duration(cfg.Cache.Modules.Conversation) * time.Second,
UnreadCountTTL: time.Duration(cfg.Cache.Modules.UnreadCount) * time.Second,
GroupMembersTTL: time.Duration(cfg.Cache.Modules.GroupMembers) * time.Second,
DisableFlushDB: cfg.Cache.DisableFlushDB,
})
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
}
cacheInstance := cache.NewRedisCache(redisClient)
zap.L().Info("Initialized Redis cache via dependency injection")
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,
userRepo repository.UserRepository,
sensitiveService service.SensitiveService,
cfg *config.Config,
) *hook.ModerationHooks {
strictMode := cfg.OpenAI.StrictModeration
moderationHooks := hook.NewModerationHooks(
postAIService,
sensitiveService,
postRepo,
commentRepo,
userRepo,
strictMode,
"***",
)
moderationHooks.RegisterAll(manager)
zap.L().Info("Registered moderation hooks",
zap.Bool("strict_mode", strictMode),
)
return moderationHooks
}
// ProvideWSMessagePublisher 提供 WebSocket 消息发布器
// standalone 模式返回 *Hubcluster 模式返回 *Bus组合 Hub + Redis Pub/Sub
func ProvideWSMessagePublisher(cfg *config.Config, redisClient *redis.Client) ws.MessagePublisher {
hub := ws.NewHub()
if cfg.WebSocket.Mode == "cluster" && redisClient != nil {
bus := ws.NewBus(hub, redisClient.GetClient(), &ws.WSClusterConfig{
InstanceID: cfg.WebSocket.Cluster.InstanceID,
MsgChannel: cfg.WebSocket.Cluster.MsgChannel,
OnlineTTL: cfg.WebSocket.Cluster.OnlineTTL,
HeartbeatInterval: cfg.WebSocket.Cluster.HeartbeatInterval,
})
if err := bus.Start(); err != nil {
zap.L().Error("Failed to start WebSocket Bus, falling back to standalone mode",
zap.Error(err),
)
return hub
}
zap.L().Info("WebSocket running in cluster mode",
zap.String("instance_id", bus.InstanceID()),
)
return bus
}
zap.L().Info("WebSocket running in standalone mode")
return hub
}
// ProvideOpenAIClient 提供 OpenAI 客户端
func ProvideOpenAIClient(cfg *config.Config) openai.Client {
return openai.NewClient(openai.ConfigFromAppConfig(&cfg.OpenAI))
}
// ProvideTencentClient 提供腾讯云TMS客户端
func ProvideTencentClient(cfg *config.Config) tencent.Client {
return tencent.NewClient(tencent.Config{
Enabled: cfg.TencentTMS.Enabled,
SecretID: cfg.TencentTMS.SecretID,
SecretKey: cfg.TencentTMS.SecretKey,
Region: cfg.TencentTMS.Region,
BizType: cfg.TencentTMS.BizType,
Timeout: cfg.TencentTMS.Timeout,
})
}
// ProvideEmailClient 提供邮件客户端
func ProvideEmailClient(cfg *config.Config) email.Client {
return email.NewClient(email.ConfigFromAppConfig(&cfg.Email))
}
// ProvideS3Client 提供 S3 客户端
func ProvideS3Client(cfg *config.Config) (*s3.Client, error) {
return s3.New(&cfg.S3)
}
// ProvideJPushClient 提供极光推送客户端
func ProvideJPushClient(cfg *config.Config, logger *zap.Logger) *jpush.Client {
if !cfg.JPush.Enabled {
logger.Info("JPush is disabled")
return nil
}
if cfg.JPush.AppKey == "" || cfg.JPush.MasterSecret == "" {
logger.Warn("JPush is enabled but app_key or master_secret is not configured")
return nil
}
client := jpush.NewClient(cfg.JPush.AppKey, cfg.JPush.MasterSecret, cfg.JPush.Production, logger)
logger.Info("JPush client initialized",
zap.Bool("production", cfg.JPush.Production),
)
return client
}
// ProvideTransactionManager 提供事务管理器
func ProvideTransactionManager(db *gorm.DB) repository.TransactionManager {
return repository.NewTransactionManager(db)
}
// ProvideMessageEncryptor 提供消息加密器
func ProvideMessageEncryptor(cfg *config.Config) error {
if !cfg.Encryption.Enabled {
zap.L().Info("Message encryption is disabled")
return nil
}
if len(cfg.Encryption.Key) != 32 {
zap.L().Error("Encryption key must be 32 bytes for AES-256",
zap.Int("actual_length", len(cfg.Encryption.Key)),
)
return nil
}
if err := crypto.InitMessageEncryptor(cfg.Encryption.Key, cfg.Encryption.KeyVersion); err != nil {
zap.L().Error("Failed to initialize message encryptor", zap.Error(err))
return err
}
zap.L().Info("Message encryption initialized successfully",
zap.Int("key_version", cfg.Encryption.KeyVersion),
)
return nil
}
// ProvideLogger 提供 Zap Logger
func ProvideLogger() *zap.Logger {
logger, err := zap.NewProduction()
if err != nil {
panic("Failed to create logger: " + err.Error())
}
// 设置为全局 logger以便在其他地方使用 zap.L()
zap.ReplaceGlobals(logger)
return logger
}