diff --git a/go.mod b/go.mod index 4af0695db..d35252536 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 github.com/mitchellh/mapstructure v1.5.0 github.com/openimsdk/protocol v0.0.73-alpha.6 - github.com/openimsdk/tools v0.0.50-alpha.81 + github.com/openimsdk/tools v0.0.50-alpha.83 github.com/pkg/errors v0.9.1 // indirect github.com/prometheus/client_golang v1.18.0 github.com/stretchr/testify v1.9.0 diff --git a/go.sum b/go.sum index 6bc410a2d..35fc3d4c6 100644 --- a/go.sum +++ b/go.sum @@ -349,8 +349,8 @@ github.com/openimsdk/gomake v0.0.15-alpha.5 h1:eEZCEHm+NsmcO3onXZPIUbGFCYPYbsX5b github.com/openimsdk/gomake v0.0.15-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI= github.com/openimsdk/protocol v0.0.73-alpha.6 h1:sna9coWG7HN1zObBPtvG0Ki/vzqHXiB4qKbA5P3w7kc= github.com/openimsdk/protocol v0.0.73-alpha.6/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw= -github.com/openimsdk/tools v0.0.50-alpha.81 h1:VbuJKtigNXLkCKB/Q6f2UHsqoSaTOAwS8F51c1nhOCA= -github.com/openimsdk/tools v0.0.50-alpha.81/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo= +github.com/openimsdk/tools v0.0.50-alpha.83 h1:7c1D40YGqIWUmGfCII5pduETGC/8c2DyS9SQ4LvoplU= +github.com/openimsdk/tools v0.0.50-alpha.83/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo= github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM= github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs= github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ= diff --git a/internal/api/router.go b/internal/api/router.go index 7c2c78c7d..add8ef36b 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -9,12 +9,8 @@ import ( "github.com/gin-gonic/gin" "github.com/gin-gonic/gin/binding" "github.com/go-playground/validator/v10" - "github.com/openimsdk/open-im-server/v3/pkg/authverify" - "github.com/openimsdk/tools/mcontext" - "github.com/openimsdk/tools/utils/datautil" - clientv3 "go.etcd.io/etcd/client/v3" - "github.com/openimsdk/open-im-server/v3/internal/api/jssdk" + "github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" "github.com/openimsdk/open-im-server/v3/pkg/common/servererrs" @@ -32,6 +28,7 @@ import ( "github.com/openimsdk/tools/discovery/etcd" "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/mw" + clientv3 "go.etcd.io/etcd/client/v3" ) const ( @@ -271,7 +268,6 @@ func newGinRouter(ctx context.Context, client discovery.Conn, cfg *Config) (*gin conversationGroup.POST("/get_owner_conversation", c.GetOwnerConversation) conversationGroup.POST("/get_not_notify_conversation_ids", c.GetNotNotifyConversationIDs) conversationGroup.POST("/get_pinned_conversation_ids", c.GetPinnedConversationIDs) - conversationGroup.POST("/update_conversations_by_user", c.UpdateConversationsByUser) } { @@ -315,7 +311,6 @@ func newGinRouter(ctx context.Context, client discovery.Conn, cfg *Config) (*gin configGroup.POST("/get_config_list", cm.GetConfigList) configGroup.POST("/get_config", cm.GetConfig) configGroup.POST("/set_config", cm.SetConfig) - configGroup.POST("/set_configs", cm.SetConfigs) configGroup.POST("/reset_config", cm.ResetConfig) configGroup.POST("/set_enable_config_manager", cm.SetEnableConfigManager) configGroup.POST("/get_enable_config_manager", cm.GetEnableConfigManager) @@ -359,9 +354,7 @@ func GinParseToken(authClient *rpcli.AuthClient) gin.HandlerFunc { func setGinIsAdmin(imAdminUserID []string) gin.HandlerFunc { return func(c *gin.Context) { - opUserID := mcontext.GetOpUserID(c) - admin := datautil.Contain(opUserID, imAdminUserID...) - c.Set(authverify.CtxIsAdminKey, admin) + c.Set(authverify.CtxAdminUserIDsKey, imAdminUserID) } } diff --git a/internal/msggateway/ws_server.go b/internal/msggateway/ws_server.go index d69062bda..211482cb1 100644 --- a/internal/msggateway/ws_server.go +++ b/internal/msggateway/ws_server.go @@ -49,7 +49,7 @@ type WsServer struct { unregisterChan chan *Client kickHandlerChan chan *kickHandler clients UserMap - online *rpccache.OnlineCache + online rpccache.OnlineCache subscription *Subscription clientPool sync.Pool onlineUserNum atomic.Int64 diff --git a/internal/msgtransfer/online_msg_to_mongo_handler.go b/internal/msgtransfer/online_msg_to_mongo_handler.go index ae14d02a1..a895bb9c4 100644 --- a/internal/msgtransfer/online_msg_to_mongo_handler.go +++ b/internal/msgtransfer/online_msg_to_mongo_handler.go @@ -48,15 +48,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(ctx context.Cont log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String()) err = mc.msgTransferDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq) if err != nil { - log.ZError( - ctx, - "single data insert to mongo err", - err, - "msg", - msgFromMQ.MsgData, - "conversationID", - msgFromMQ.ConversationID, - ) + log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID) prommetrics.MsgInsertMongoFailedCounter.Inc() } else { prommetrics.MsgInsertMongoSuccessCounter.Inc() diff --git a/internal/push/push_handler.go b/internal/push/push_handler.go index 418c4c7f2..7b1efe3bf 100644 --- a/internal/push/push_handler.go +++ b/internal/push/push_handler.go @@ -33,7 +33,7 @@ type ConsumerHandler struct { offlinePusher offlinepush.OfflinePusher onlinePusher OnlinePusher pushDatabase controller.PushDatabase - onlineCache *rpccache.OnlineCache + onlineCache rpccache.OnlineCache groupLocalCache *rpccache.GroupLocalCache conversationLocalCache *rpccache.ConversationLocalCache webhookClient *webhook.Client @@ -120,11 +120,7 @@ func (c *ConsumerHandler) HandleMs2PsChat(ctx context.Context, msg []byte) { } func (c *ConsumerHandler) WaitCache() { - c.onlineCache.Lock.Lock() - for c.onlineCache.CurrentPhase.Load() < rpccache.DoSubscribeOver { - c.onlineCache.Cond.Wait() - } - c.onlineCache.Lock.Unlock() + c.onlineCache.WaitCache() } // Push2User Suitable for two types of conversations, one is SingleChatType and the other is NotificationChatType. diff --git a/pkg/authverify/token.go b/pkg/authverify/token.go index 75fb1448b..2e3639776 100644 --- a/pkg/authverify/token.go +++ b/pkg/authverify/token.go @@ -51,15 +51,15 @@ func CheckSystemAccount(ctx context.Context, level int32) bool { } const ( - CtxIsAdminKey = "CtxIsAdminKey" + CtxAdminUserIDsKey = "CtxAdminUserIDsKey" ) func WithIMAdminUserIDs(ctx context.Context, imAdminUserID []string) context.Context { - return context.WithValue(ctx, CtxIsAdminKey, imAdminUserID) + return context.WithValue(ctx, CtxAdminUserIDsKey, imAdminUserID) } func GetIMAdminUserIDs(ctx context.Context) []string { - imAdminUserID, _ := ctx.Value(CtxIsAdminKey).([]string) + imAdminUserID, _ := ctx.Value(CtxAdminUserIDsKey).([]string) return imAdminUserID } diff --git a/pkg/common/startrpc/start.go b/pkg/common/startrpc/start.go index 03621343b..b99d32db1 100644 --- a/pkg/common/startrpc/start.go +++ b/pkg/common/startrpc/start.go @@ -47,64 +47,6 @@ func init() { prommetrics.RegistryAll() } -func getConfigRpcMaxRequestBody(value reflect.Value) *conf.MaxRequestBody { - for value.Kind() == reflect.Pointer { - value = value.Elem() - } - if value.Kind() == reflect.Struct { - num := value.NumField() - for i := 0; i < num; i++ { - field := value.Field(i) - if !field.CanInterface() { - continue - } - for field.Kind() == reflect.Pointer { - field = field.Elem() - } - switch elem := field.Interface().(type) { - case conf.Share: - return &elem.RPCMaxBodySize - case conf.MaxRequestBody: - return &elem - } - if field.Kind() == reflect.Struct { - if elem := getConfigRpcMaxRequestBody(field); elem != nil { - return elem - } - } - } - } - return nil -} - -func getConfigShare(value reflect.Value) *conf.Share { - for value.Kind() == reflect.Pointer { - value = value.Elem() - } - if value.Kind() == reflect.Struct { - num := value.NumField() - for i := 0; i < num; i++ { - field := value.Field(i) - if !field.CanInterface() { - continue - } - for field.Kind() == reflect.Pointer { - field = field.Elem() - } - switch elem := field.Interface().(type) { - case conf.Share: - return &elem - } - if field.Kind() == reflect.Struct { - if elem := getConfigShare(field); elem != nil { - return elem - } - } - } - } - return nil -} - func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP, registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, notification *conf.Notification, config T, watchConfigNames []string, watchServiceNames []string, @@ -122,8 +64,8 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c options = append(options, grpcsrv.GrpcServerMetadataContext(), - grpcsrv.GrpcServerLogger(), grpcsrv.GrpcServerErrorConvert(), + grpcsrv.GrpcServerLogger(), grpcsrv.GrpcServerRequestValidate(), grpcsrv.GrpcServerPanicCapture(), ) diff --git a/pkg/common/startrpc/tools.go b/pkg/common/startrpc/tools.go new file mode 100644 index 000000000..54b518181 --- /dev/null +++ b/pkg/common/startrpc/tools.go @@ -0,0 +1,39 @@ +package startrpc + +import ( + "reflect" + + conf "github.com/openimsdk/open-im-server/v3/pkg/common/config" +) + +func getConfig[T any](value reflect.Value) *T { + for value.Kind() == reflect.Pointer { + value = value.Elem() + } + if value.Kind() == reflect.Struct { + num := value.NumField() + for i := 0; i < num; i++ { + field := value.Field(i) + for field.Kind() == reflect.Pointer { + field = field.Elem() + } + if field.Kind() == reflect.Struct { + if elem, ok := field.Interface().(T); ok { + return &elem + } + if elem := getConfig[T](field); elem != nil { + return elem + } + } + } + } + return nil +} + +func getConfigRpcMaxRequestBody(value reflect.Value) *conf.MaxRequestBody { + return getConfig[conf.MaxRequestBody](value) +} + +func getConfigShare(value reflect.Value) *conf.Share { + return getConfig[conf.Share](value) +} diff --git a/pkg/rpccache/online.go b/pkg/rpccache/online.go index 959b054a2..9b8c03101 100644 --- a/pkg/rpccache/online.go +++ b/pkg/rpccache/online.go @@ -9,6 +9,7 @@ import ( "sync/atomic" "time" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/rpcli" "github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/user" @@ -23,9 +24,25 @@ import ( "github.com/redis/go-redis/v9" ) -func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (*OnlineCache, error) { +const ( + Begin uint32 = iota + DoOnlineStatusOver + DoSubscribeOver +) + +type OnlineCache interface { + GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) + GetUserOnline(ctx context.Context, userID string) (bool, error) + GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) + WaitCache() +} + +func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (OnlineCache, error) { + if config.Standalone() { + return disableOnlineCache{client: client}, nil + } l := &sync.Mutex{} - x := &OnlineCache{ + x := &defaultOnlineCache{ client: client, group: group, fullUserCache: fullUserCache, @@ -60,13 +77,7 @@ func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis. return x, nil } -const ( - Begin uint32 = iota - DoOnlineStatusOver - DoSubscribeOver -) - -type OnlineCache struct { +type defaultOnlineCache struct { client *rpcli.UserClient group *GroupLocalCache @@ -82,7 +93,7 @@ type OnlineCache struct { CurrentPhase atomic.Uint32 } -func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) { +func (o *defaultOnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) { log.ZDebug(ctx, "init users online status begin") var ( @@ -135,7 +146,7 @@ func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) { return nil } -func (o *OnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) { +func (o *defaultOnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) { o.Lock.Lock() ch := rdb.Subscribe(ctx, cachekey.OnlineChannel).Channel() for o.CurrentPhase.Load() < DoOnlineStatusOver { @@ -186,7 +197,7 @@ func (o *OnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient } } -func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) { +func (o *defaultOnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) { platformIDs, err := o.lruCache.Get(userID, func() ([]int32, error) { return o.client.GetUserOnlinePlatform(ctx, userID) }) @@ -198,7 +209,7 @@ func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) return platformIDs, nil } -func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) { +func (o *defaultOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) { platformIDs, err := o.getUserOnlinePlatform(ctx, userID) if err != nil { return nil, err @@ -208,7 +219,7 @@ func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) return platformIDs, nil } -func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) { +func (o *defaultOnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) { platformIDs, err := o.getUserOnlinePlatform(ctx, userID) if err != nil { return false, err @@ -216,7 +227,7 @@ func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, e return len(platformIDs) > 0, nil } -func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) { +func (o *defaultOnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) { if len(userIDs) == 0 { return nil, nil } @@ -240,7 +251,7 @@ func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs [] return platformIDsMap, nil } -func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) { +func (o *defaultOnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) { t := time.Now() var ( @@ -276,7 +287,7 @@ func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]s return onlineUserIDs, offlineUserIDs, nil } -func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) { +func (o *defaultOnlineCache) setUserOnline(userID string, platformIDs []int32) { switch o.fullUserCache { case true: o.mapCache.Store(userID, platformIDs) @@ -285,6 +296,51 @@ func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) { } } -func (o *OnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool { +func (o *defaultOnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool { return o.lruCache.SetHas(userID, platformIDs) } + +func (o *defaultOnlineCache) WaitCache() { + o.Lock.Lock() + for o.CurrentPhase.Load() < DoSubscribeOver { + o.Cond.Wait() + } + o.Lock.Unlock() +} + +type disableOnlineCache struct { + client *rpcli.UserClient +} + +func (o disableOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) { + return o.client.GetUserOnlinePlatform(ctx, userID) +} + +func (o disableOnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) { + onlinePlatform, err := o.client.GetUserOnlinePlatform(ctx, userID) + if err != nil { + return false, err + } + return len(onlinePlatform) > 0, err +} + +func (o disableOnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) { + var ( + onlineUserIDs = make([]string, 0, len(userIDs)) + offlineUserIDs = make([]string, 0, len(userIDs)) + ) + for _, userID := range userIDs { + online, err := o.GetUserOnline(ctx, userID) + if err != nil { + return nil, nil, err + } + if online { + onlineUserIDs = append(onlineUserIDs, userID) + } else { + offlineUserIDs = append(offlineUserIDs, userID) + } + } + return onlineUserIDs, offlineUserIDs, nil +} + +func (o disableOnlineCache) WaitCache() {}