Merge branch 'openimsdk:main' into main

internal/api/config_manager.go

@@ -15,6 +15,7 @@ import (
 	"github.com/openimsdk/tools/apiresp"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/openimsdk/tools/utils/runtimeenv"
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
@@ -144,6 +145,110 @@ func (cm *ConfigManager) SetConfig(c *gin.Context) {
 	apiresp.GinSuccess(c, nil)
 }
 
+func (cm *ConfigManager) SetConfigs(c *gin.Context) {
+	if cm.config.Discovery.Enable != config.ETCD {
+		apiresp.GinError(c, errs.New("only etcd support set config").Wrap())
+		return
+	}
+	var req apistruct.SetConfigsReq
+	if err := c.BindJSON(&req); err != nil {
+		apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
+		return
+	}
+	var (
+		err error
+		ops []*clientv3.Op
+	)
+
+	for _, cf := range req.Configs {
+		var op *clientv3.Op
+		switch cf.ConfigName {
+		case cm.config.Discovery.GetConfigFileName():
+			op, err = compareAndOp[config.Discovery](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Kafka.GetConfigFileName():
+			op, err = compareAndOp[config.Kafka](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.LocalCache.GetConfigFileName():
+			op, err = compareAndOp[config.LocalCache](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Log.GetConfigFileName():
+			op, err = compareAndOp[config.Log](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Minio.GetConfigFileName():
+			op, err = compareAndOp[config.Minio](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Mongo.GetConfigFileName():
+			op, err = compareAndOp[config.Mongo](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Notification.GetConfigFileName():
+			op, err = compareAndOp[config.Notification](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.API.GetConfigFileName():
+			op, err = compareAndOp[config.API](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.CronTask.GetConfigFileName():
+			op, err = compareAndOp[config.CronTask](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.MsgGateway.GetConfigFileName():
+			op, err = compareAndOp[config.MsgGateway](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.MsgTransfer.GetConfigFileName():
+			op, err = compareAndOp[config.MsgTransfer](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Push.GetConfigFileName():
+			op, err = compareAndOp[config.Push](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Auth.GetConfigFileName():
+			op, err = compareAndOp[config.Auth](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Conversation.GetConfigFileName():
+			op, err = compareAndOp[config.Conversation](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Friend.GetConfigFileName():
+			op, err = compareAndOp[config.Friend](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Group.GetConfigFileName():
+			op, err = compareAndOp[config.Group](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Msg.GetConfigFileName():
+			op, err = compareAndOp[config.Msg](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Third.GetConfigFileName():
+			op, err = compareAndOp[config.Third](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.User.GetConfigFileName():
+			op, err = compareAndOp[config.User](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Redis.GetConfigFileName():
+			op, err = compareAndOp[config.Redis](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Share.GetConfigFileName():
+			op, err = compareAndOp[config.Share](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		case cm.config.Webhooks.GetConfigFileName():
+			op, err = compareAndOp[config.Webhooks](c, cm.config.Name2Config(cf.ConfigName), &cf, cm)
+		default:
+			apiresp.GinError(c, errs.ErrArgs.Wrap())
+			return
+		}
+		if err != nil {
+			apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
+			return
+		}
+		if op != nil {
+			ops = append(ops, op)
+		}
+	}
+	if len(ops) > 0 {
+		tx := cm.client.Txn(c)
+		if _, err = tx.Then(datautil.Batch(func(op *clientv3.Op) clientv3.Op { return *op }, ops)...).Commit(); err != nil {
+			apiresp.GinError(c, errs.WrapMsg(err, "save to etcd failed"))
+			return
+		}
+
+	}
+
+	apiresp.GinSuccess(c, nil)
+}
+
+func compareAndOp[T any](c *gin.Context, old any, req *apistruct.SetConfigReq, cm *ConfigManager) (*clientv3.Op, error) {
+	conf := new(T)
+	err := json.Unmarshal([]byte(req.Data), &conf)
+	if err != nil {
+		return nil, errs.ErrArgs.WithDetail(err.Error()).Wrap()
+	}
+	eq := reflect.DeepEqual(old, conf)
+	if eq {
+		return nil, nil
+	}
+	data, err := json.Marshal(conf)
+	if err != nil {
+		return nil, errs.ErrArgs.WithDetail(err.Error()).Wrap()
+	}
+	op := clientv3.OpPut(etcd.BuildKey(req.ConfigName), string(data))
+	return &op, nil
+}
+
 func compareAndSave[T any](c *gin.Context, old any, req *apistruct.SetConfigReq, cm *ConfigManager) error {
 	conf := new(T)
 	err := json.Unmarshal([]byte(req.Data), &conf)

internal/api/router.go

@@ -307,6 +307,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)

internal/rpc/conversation/conversation.go

@@ -237,6 +237,7 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver
 	if req.Conversation == nil {
 		return nil, errs.ErrArgs.WrapMsg("conversation must not be nil")
 	}
+
 	if req.Conversation.ConversationType == constant.WriteGroupChatType {
 		groupInfo, err := c.groupClient.GetGroupInfo(ctx, req.Conversation.GroupID)
 		if err != nil {
@@ -271,109 +272,29 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver
 	conversation.UserID = req.Conversation.UserID
 	conversation.GroupID = req.Conversation.GroupID
 
-	m := make(map[string]any)
-
-	setConversationFieldsFunc := func() {
-		if req.Conversation.RecvMsgOpt != nil {
-			conversation.RecvMsgOpt = req.Conversation.RecvMsgOpt.Value
-			m["recv_msg_opt"] = req.Conversation.RecvMsgOpt.Value
-		}
-		if req.Conversation.AttachedInfo != nil {
-			conversation.AttachedInfo = req.Conversation.AttachedInfo.Value
-			m["attached_info"] = req.Conversation.AttachedInfo.Value
-		}
-		if req.Conversation.Ex != nil {
-			conversation.Ex = req.Conversation.Ex.Value
-			m["ex"] = req.Conversation.Ex.Value
-		}
-		if req.Conversation.IsPinned != nil {
-			conversation.IsPinned = req.Conversation.IsPinned.Value
-			m["is_pinned"] = req.Conversation.IsPinned.Value
-		}
-		if req.Conversation.GroupAtType != nil {
-			conversation.GroupAtType = req.Conversation.GroupAtType.Value
-			m["group_at_type"] = req.Conversation.GroupAtType.Value
-		}
-		if req.Conversation.MsgDestructTime != nil {
-			conversation.MsgDestructTime = req.Conversation.MsgDestructTime.Value
-			m["msg_destruct_time"] = req.Conversation.MsgDestructTime.Value
-		}
-		if req.Conversation.IsMsgDestruct != nil {
-			conversation.IsMsgDestruct = req.Conversation.IsMsgDestruct.Value
-			m["is_msg_destruct"] = req.Conversation.IsMsgDestruct.Value
-		}
-		if req.Conversation.BurnDuration != nil {
-			conversation.BurnDuration = req.Conversation.BurnDuration.Value
-			m["burn_duration"] = req.Conversation.BurnDuration.Value
-		}
+	m, conversation, err := UpdateConversationsMap(ctx, req)
+	if err != nil {
+		return nil, err
 	}
 
-	// set need set field in conversation
-	setConversationFieldsFunc()
-
 	for userID := range conversationMap {
-		unequal := len(m)
+		unequal := UserUpdateCheckMap(ctx, userID, req.Conversation, conversationMap[userID])
 
-		if req.Conversation.RecvMsgOpt != nil {
-			if req.Conversation.RecvMsgOpt.Value == conversationMap[userID].RecvMsgOpt {
-				unequal--
-			}
-		}
-
-		if req.Conversation.AttachedInfo != nil {
-			if req.Conversation.AttachedInfo.Value == conversationMap[userID].AttachedInfo {
-				unequal--
-			}
-		}
-
-		if req.Conversation.Ex != nil {
-			if req.Conversation.Ex.Value == conversationMap[userID].Ex {
-				unequal--
-			}
-		}
-		if req.Conversation.IsPinned != nil {
-			if req.Conversation.IsPinned.Value == conversationMap[userID].IsPinned {
-				unequal--
-			}
-		}
-
-		if req.Conversation.GroupAtType != nil {
-			if req.Conversation.GroupAtType.Value == conversationMap[userID].GroupAtType {
-				unequal--
-			}
-		}
-
-		if req.Conversation.MsgDestructTime != nil {
-			if req.Conversation.MsgDestructTime.Value == conversationMap[userID].MsgDestructTime {
-				unequal--
-			}
-		}
-
-		if req.Conversation.IsMsgDestruct != nil {
-			if req.Conversation.IsMsgDestruct.Value == conversationMap[userID].IsMsgDestruct {
-				unequal--
-			}
-		}
-
-		if req.Conversation.BurnDuration != nil {
-			if req.Conversation.BurnDuration.Value == conversationMap[userID].BurnDuration {
-				unequal--
-			}
-		}
-
-		if unequal > 0 {
+		if unequal {
 			needUpdateUsersList = append(needUpdateUsersList, userID)
 		}
 	}
+
 	if len(m) != 0 && len(needUpdateUsersList) != 0 {
 		if err := c.conversationDatabase.SetUsersConversationFieldTx(ctx, needUpdateUsersList, &conversation, m); err != nil {
 			return nil, err
 		}
 
-		for _, v := range needUpdateUsersList {
-			c.conversationNotificationSender.ConversationChangeNotification(ctx, v, []string{req.Conversation.ConversationID})
+		for _, userID := range needUpdateUsersList {
+			c.conversationNotificationSender.ConversationChangeNotification(ctx, userID, []string{req.Conversation.ConversationID})
 		}
 	}
+
 	if req.Conversation.IsPrivateChat != nil && req.Conversation.ConversationType != constant.ReadGroupChatType {
 		var conversations []*dbModel.Conversation
 		for _, ownerUserID := range req.UserIDs {

internal/rpc/conversation/db_map.go

@@ -0,0 +1,85 @@
+package conversation
+
+import (
+	"context"
+
+	dbModel "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"github.com/openimsdk/protocol/conversation"
+)
+
+func UpdateConversationsMap(ctx context.Context, req *conversation.SetConversationsReq) (m map[string]any, conversation dbModel.Conversation, err error) {
+	m = make(map[string]any)
+
+	conversation.ConversationID = req.Conversation.ConversationID
+	conversation.ConversationType = req.Conversation.ConversationType
+	conversation.UserID = req.Conversation.UserID
+	conversation.GroupID = req.Conversation.GroupID
+
+	if req.Conversation.RecvMsgOpt != nil {
+		conversation.RecvMsgOpt = req.Conversation.RecvMsgOpt.Value
+		m["recv_msg_opt"] = req.Conversation.RecvMsgOpt.Value
+	}
+
+	if req.Conversation.AttachedInfo != nil {
+		conversation.AttachedInfo = req.Conversation.AttachedInfo.Value
+		m["attached_info"] = req.Conversation.AttachedInfo.Value
+	}
+
+	if req.Conversation.Ex != nil {
+		conversation.Ex = req.Conversation.Ex.Value
+		m["ex"] = req.Conversation.Ex.Value
+	}
+	if req.Conversation.IsPinned != nil {
+		conversation.IsPinned = req.Conversation.IsPinned.Value
+		m["is_pinned"] = req.Conversation.IsPinned.Value
+	}
+	if req.Conversation.GroupAtType != nil {
+		conversation.GroupAtType = req.Conversation.GroupAtType.Value
+		m["group_at_type"] = req.Conversation.GroupAtType.Value
+	}
+	if req.Conversation.MsgDestructTime != nil {
+		conversation.MsgDestructTime = req.Conversation.MsgDestructTime.Value
+		m["msg_destruct_time"] = req.Conversation.MsgDestructTime.Value
+	}
+	if req.Conversation.IsMsgDestruct != nil {
+		conversation.IsMsgDestruct = req.Conversation.IsMsgDestruct.Value
+		m["is_msg_destruct"] = req.Conversation.IsMsgDestruct.Value
+	}
+	if req.Conversation.BurnDuration != nil {
+		conversation.BurnDuration = req.Conversation.BurnDuration.Value
+		m["burn_duration"] = req.Conversation.BurnDuration.Value
+	}
+
+	return m, conversation, nil
+}
+
+func UserUpdateCheckMap(ctx context.Context, userID string, req *conversation.ConversationReq, conversation *dbModel.Conversation) (unequal bool) {
+	unequal = false
+
+	if req.RecvMsgOpt != nil && conversation.RecvMsgOpt != req.RecvMsgOpt.Value {
+		unequal = true
+	}
+	if req.AttachedInfo != nil && conversation.AttachedInfo != req.AttachedInfo.Value {
+		unequal = true
+	}
+	if req.Ex != nil && conversation.Ex != req.Ex.Value {
+		unequal = true
+	}
+	if req.IsPinned != nil && conversation.IsPinned != req.IsPinned.Value {
+		unequal = true
+	}
+	if req.GroupAtType != nil && conversation.GroupAtType != req.GroupAtType.Value {
+		unequal = true
+	}
+	if req.MsgDestructTime != nil && conversation.MsgDestructTime != req.MsgDestructTime.Value {
+		unequal = true
+	}
+	if req.IsMsgDestruct != nil && conversation.IsMsgDestruct != req.IsMsgDestruct.Value {
+		unequal = true
+	}
+	if req.BurnDuration != nil && conversation.BurnDuration != req.BurnDuration.Value {
+		unequal = true
+	}
+
+	return unequal
+}

internal/rpc/group/group.go

@@ -390,6 +390,8 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 		if err := g.PopulateGroupMember(ctx, groupMember); err != nil {
 			return nil, err
 		}
+	} else {
+		opUserID = mcontext.GetOpUserID(ctx)
 	}
 
 	if err := g.webhookBeforeInviteUserToGroup(ctx, &g.config.WebhooksConfig.BeforeInviteUserToGroup, req); err != nil && err != servererrs.ErrCallbackContinue {
@@ -425,6 +427,7 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 			}
 		}
 	}
+
 	var groupMembers []*model.GroupMember
 	for _, userID := range req.InvitedUserIDs {
 		member := &model.GroupMember{

pkg/apistruct/config_manager.go

@@ -15,6 +15,10 @@ type SetConfigReq struct {
 	Data       string `json:"data"`
 }
 
+type SetConfigsReq struct {
+	Configs []SetConfigReq `json:"configs"`
+}
+
 type SetEnableConfigManagerReq struct {
 	Enable bool `json:"enable"`
 }