Merge 5556e89fa3d46a521247c755de0befcd839c17d9 into 1d7660bedb98c692cdc1f09d84428acaf9cb7ea2

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.83
+	github.com/openimsdk/tools v0.0.50-alpha.81
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0

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.83 h1:7c1D40YGqIWUmGfCII5pduETGC/8c2DyS9SQ4LvoplU=
-github.com/openimsdk/tools v0.0.50-alpha.83/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
+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/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=

internal/api/router.go

@@ -9,8 +9,12 @@ 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/internal/api/jssdk"
 	"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/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
@@ -28,7 +32,6 @@ 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 (
@@ -268,6 +271,7 @@ 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)
 	}
 
 	{
@@ -311,6 +315,7 @@ 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)
@@ -354,7 +359,9 @@ func GinParseToken(authClient *rpcli.AuthClient) gin.HandlerFunc {
 
 func setGinIsAdmin(imAdminUserID []string) gin.HandlerFunc {
 	return func(c *gin.Context) {
-		c.Set(authverify.CtxAdminUserIDsKey, imAdminUserID)
+		opUserID := mcontext.GetOpUserID(c)
+		admin := datautil.Contain(opUserID, imAdminUserID...)
+		c.Set(authverify.CtxIsAdminKey, admin)
 	}
 }
 

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

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -48,7 +48,15 @@ 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()

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,7 +120,11 @@ func (c *ConsumerHandler) HandleMs2PsChat(ctx context.Context, msg []byte) {
 }
 
 func (c *ConsumerHandler) WaitCache() {
-	c.onlineCache.WaitCache()
+	c.onlineCache.Lock.Lock()
+	for c.onlineCache.CurrentPhase.Load() < rpccache.DoSubscribeOver {
+		c.onlineCache.Cond.Wait()
+	}
+	c.onlineCache.Lock.Unlock()
 }
 
 // Push2User Suitable for two types of conversations, one is SingleChatType and the other is NotificationChatType.

pkg/authverify/token.go

@@ -51,15 +51,15 @@ func CheckSystemAccount(ctx context.Context, level int32) bool {
 }
 
 const (
-	CtxAdminUserIDsKey = "CtxAdminUserIDsKey"
+	CtxIsAdminKey = "CtxIsAdminKey"
 )
 
 func WithIMAdminUserIDs(ctx context.Context, imAdminUserID []string) context.Context {
-	return context.WithValue(ctx, CtxAdminUserIDsKey, imAdminUserID)
+	return context.WithValue(ctx, CtxIsAdminKey, imAdminUserID)
 }
 
 func GetIMAdminUserIDs(ctx context.Context) []string {
-	imAdminUserID, _ := ctx.Value(CtxAdminUserIDsKey).([]string)
+	imAdminUserID, _ := ctx.Value(CtxIsAdminKey).([]string)
 	return imAdminUserID
 }
 

pkg/common/startrpc/start.go

@@ -47,6 +47,64 @@ 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,
@@ -64,8 +122,8 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
 
 	options = append(options,
 		grpcsrv.GrpcServerMetadataContext(),
-		grpcsrv.GrpcServerErrorConvert(),
 		grpcsrv.GrpcServerLogger(),
+		grpcsrv.GrpcServerErrorConvert(),
 		grpcsrv.GrpcServerRequestValidate(),
 		grpcsrv.GrpcServerPanicCapture(),
 	)

pkg/common/startrpc/tools.go

@@ -1,39 +0,0 @@
-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)
-}

pkg/rpccache/online.go

@@ -9,7 +9,6 @@ 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"
@@ -24,25 +23,9 @@ import (
 	"github.com/redis/go-redis/v9"
 )
 
-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
-	}
+func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (*OnlineCache, error) {
 	l := &sync.Mutex{}
-	x := &defaultOnlineCache{
+	x := &OnlineCache{
 		client:        client,
 		group:         group,
 		fullUserCache: fullUserCache,
@@ -77,7 +60,13 @@ func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.
 	return x, nil
 }
 
-type defaultOnlineCache struct {
+const (
+	Begin uint32 = iota
+	DoOnlineStatusOver
+	DoSubscribeOver
+)
+
+type OnlineCache struct {
 	client *rpcli.UserClient
 	group  *GroupLocalCache
 
@@ -93,7 +82,7 @@ type defaultOnlineCache struct {
 	CurrentPhase atomic.Uint32
 }
 
-func (o *defaultOnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
+func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
 	log.ZDebug(ctx, "init users online status begin")
 
 	var (
@@ -146,7 +135,7 @@ func (o *defaultOnlineCache) initUsersOnlineStatus(ctx context.Context) (err err
 	return nil
 }
 
-func (o *defaultOnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) {
+func (o *OnlineCache) 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 {
@@ -197,7 +186,7 @@ func (o *defaultOnlineCache) doSubscribe(ctx context.Context, rdb redis.Universa
 	}
 }
 
-func (o *defaultOnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
+func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
 	platformIDs, err := o.lruCache.Get(userID, func() ([]int32, error) {
 		return o.client.GetUserOnlinePlatform(ctx, userID)
 	})
@@ -209,7 +198,7 @@ func (o *defaultOnlineCache) getUserOnlinePlatform(ctx context.Context, userID s
 	return platformIDs, nil
 }
 
-func (o *defaultOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
+func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
 	platformIDs, err := o.getUserOnlinePlatform(ctx, userID)
 	if err != nil {
 		return nil, err
@@ -219,7 +208,7 @@ func (o *defaultOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID s
 	return platformIDs, nil
 }
 
-func (o *defaultOnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
+func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
 	platformIDs, err := o.getUserOnlinePlatform(ctx, userID)
 	if err != nil {
 		return false, err
@@ -227,7 +216,7 @@ func (o *defaultOnlineCache) GetUserOnline(ctx context.Context, userID string) (
 	return len(platformIDs) > 0, nil
 }
 
-func (o *defaultOnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
+func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
 	if len(userIDs) == 0 {
 		return nil, nil
 	}
@@ -251,7 +240,7 @@ func (o *defaultOnlineCache) getUserOnlinePlatformBatch(ctx context.Context, use
 	return platformIDsMap, nil
 }
 
-func (o *defaultOnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
+func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
 	t := time.Now()
 
 	var (
@@ -287,7 +276,7 @@ func (o *defaultOnlineCache) GetUsersOnline(ctx context.Context, userIDs []strin
 	return onlineUserIDs, offlineUserIDs, nil
 }
 
-func (o *defaultOnlineCache) setUserOnline(userID string, platformIDs []int32) {
+func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {
 	switch o.fullUserCache {
 	case true:
 		o.mapCache.Store(userID, platformIDs)
@@ -296,51 +285,6 @@ func (o *defaultOnlineCache) setUserOnline(userID string, platformIDs []int32) {
 	}
 }
 
-func (o *defaultOnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool {
+func (o *OnlineCache) 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() {}