merge: pre-release-v3.8.4

icey-yu · icey-yu · commit c92c8a7c12d4 · 2025-11-23T12:23:20.000+08:00
diff --git a/.github/workflows/update-version-file-on-release.yml b/.github/workflows/update-version-file-on-release.yml
@@ -96,24 +96,24 @@ jobs:
                   repo,
                   per_page: 100
                 });
-                
+            
                 release = releases.data.find(r => r.draft && r.tag_name === tagName);            
                 if (!release) {
                   throw new Error(`No release found with tag ${tagName}`);
                 }
               }
-              
+            
               await github.rest.repos.updateRelease({
                 owner,
                 repo,
                 release_id: release.id,
                 draft: false,
                 prerelease: release.prerelease
               });
-              
+            
               const status = release.draft ? "was draft" : "was already published";
               core.info(`Release ${tagName} ensured to be published (${status}).`);
-              
+            
             } catch (error) {
               core.warning(`Could not find or update release for tag ${tagName}: ${error.message}`);
             }
diff --git a/config/openim-rpc-msg.yml b/config/openim-rpc-msg.yml
@@ -1,6 +1,6 @@
 rpc:
   # The IP address where this RPC service registers itself; if left blank, it defaults to the internal network IP
-  registerIP: 
+  registerIP:
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
   # autoSetPorts indicates whether to automatically set the ports
diff --git a/go.mod b/go.mod
@@ -27,6 +27,7 @@ require (
 require github.com/google/uuid v1.6.0
 
 require (
+	github.com/IBM/sarama v1.43.0
 	github.com/fatih/color v1.14.1
 	github.com/gin-contrib/gzip v1.0.1
 	github.com/go-redis/redis v6.15.9+incompatible
@@ -54,7 +55,6 @@ require (
 	cloud.google.com/go/iam v1.1.7 // indirect
 	cloud.google.com/go/longrunning v0.5.5 // indirect
 	cloud.google.com/go/storage v1.40.0 // indirect
-	github.com/IBM/sarama v1.43.0 // indirect
 	github.com/MicahParks/keyfunc v1.9.0 // indirect
 	github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible // indirect
 	github.com/aws/aws-sdk-go-v2 v1.32.5 // indirect
diff --git a/internal/api/router.go b/internal/api/router.go
@@ -8,7 +8,6 @@ import (
 	"github.com/gin-contrib/gzip"
 	"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/open-im-server/v3/pkg/common/config"
@@ -290,6 +289,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf
 		conversationGroup.POST("/get_not_notify_conversation_ids", c.GetNotNotifyConversationIDs)
 		conversationGroup.POST("/get_pinned_conversation_ids", c.GetPinnedConversationIDs)
 		conversationGroup.POST("/delete_conversations", c.DeleteConversations)
+		conversationGroup.POST("/update_conversations_by_user", c.UpdateConversationsByUser)
 	}
 
 	{
diff --git a/internal/rpc/conversation/sync.go b/internal/rpc/conversation/sync.go
@@ -27,7 +27,7 @@ func (c *conversationServer) GetFullOwnerConversationIDs(ctx context.Context, re
 		conversationIDs = nil
 	}
 	return &conversation.GetFullOwnerConversationIDsResp{
-		Version:         idHash,
+		Version:         uint64(vl.Version),
 		VersionID:       vl.ID.Hex(),
 		Equal:           req.IdHash == idHash,
 		ConversationIDs: conversationIDs,
diff --git a/internal/rpc/group/sync.go b/internal/rpc/group/sync.go
@@ -34,7 +34,7 @@ func (g *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgrou
 		userIDs = nil
 	}
 	return &pbgroup.GetFullGroupMemberUserIDsResp{
-		Version:   idHash,
+		Version:   uint64(vl.Version),
 		VersionID: vl.ID.Hex(),
 		Equal:     req.IdHash == idHash,
 		UserIDs:   userIDs,
@@ -58,7 +58,7 @@ func (g *groupServer) GetFullJoinGroupIDs(ctx context.Context, req *pbgroup.GetF
 		groupIDs = nil
 	}
 	return &pbgroup.GetFullJoinGroupIDsResp{
-		Version:   idHash,
+		Version:   uint64(vl.Version),
 		VersionID: vl.ID.Hex(),
 		Equal:     req.IdHash == idHash,
 		GroupIDs:  groupIDs,
diff --git a/internal/rpc/relation/sync.go b/internal/rpc/relation/sync.go
@@ -56,7 +56,7 @@ func (s *friendServer) GetFullFriendUserIDs(ctx context.Context, req *relation.G
 		userIDs = nil
 	}
 	return &relation.GetFullFriendUserIDsResp{
-		Version:   idHash,
+		Version:   uint64(vl.Version),
 		VersionID: vl.ID.Hex(),
 		Equal:     req.IdHash == idHash,
 		UserIDs:   userIDs,
diff --git a/pkg/common/storage/cache/redis/token.go b/pkg/common/storage/cache/redis/token.go
@@ -13,7 +13,6 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/utils/datautil"
-	"github.com/redis/go-redis/v9"
 )
 
 type tokenCache struct {
@@ -220,7 +219,6 @@ func (c *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, p
 	if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil {
 		return errs.Wrap(err)
 	}
-
 	if c.localCache != nil {
 		c.removeLocalTokenCache(ctx, key)
 	}
diff --git a/pkg/common/storage/kafka/config.go b/pkg/common/storage/kafka/config.go
@@ -0,0 +1,33 @@
+// Copyright © 2024 OpenIM open source community. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kafka
+
+type TLSConfig struct {
+	EnableTLS          bool   `yaml:"enableTLS"`
+	CACrt              string `yaml:"caCrt"`
+	ClientCrt          string `yaml:"clientCrt"`
+	ClientKey          string `yaml:"clientKey"`
+	ClientKeyPwd       string `yaml:"clientKeyPwd"`
+	InsecureSkipVerify bool   `yaml:"insecureSkipVerify"`
+}
+
+type Config struct {
+	Username     string    `yaml:"username"`
+	Password     string    `yaml:"password"`
+	ProducerAck  string    `yaml:"producerAck"`
+	CompressType string    `yaml:"compressType"`
+	Addr         []string  `yaml:"addr"`
+	TLS          TLSConfig `yaml:"tls"`
+}
diff --git a/pkg/common/storage/kafka/consumer_group.go b/pkg/common/storage/kafka/consumer_group.go
@@ -0,0 +1,68 @@
+// Copyright © 2023 OpenIM. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kafka
+
+import (
+	"context"
+	"errors"
+
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/tools/log"
+)
+
+type MConsumerGroup struct {
+	sarama.ConsumerGroup
+	groupID string
+	topics  []string
+}
+
+func NewMConsumerGroup(conf *Config, groupID string, topics []string, autoCommitEnable bool) (*MConsumerGroup, error) {
+	config, err := BuildConsumerGroupConfig(conf, sarama.OffsetNewest, autoCommitEnable)
+	if err != nil {
+		return nil, err
+	}
+	group, err := NewConsumerGroup(config, conf.Addr, groupID)
+	if err != nil {
+		return nil, err
+	}
+	return &MConsumerGroup{
+		ConsumerGroup: group,
+		groupID:       groupID,
+		topics:        topics,
+	}, nil
+}
+
+func (mc *MConsumerGroup) GetContextFromMsg(cMsg *sarama.ConsumerMessage) context.Context {
+	return GetContextWithMQHeader(cMsg.Headers)
+}
+
+func (mc *MConsumerGroup) RegisterHandleAndConsumer(ctx context.Context, handler sarama.ConsumerGroupHandler) {
+	for {
+		err := mc.ConsumerGroup.Consume(ctx, mc.topics, handler)
+		if errors.Is(err, sarama.ErrClosedConsumerGroup) {
+			return
+		}
+		if errors.Is(err, context.Canceled) {
+			return
+		}
+		if err != nil {
+			log.ZWarn(ctx, "consume err", err, "topic", mc.topics, "groupID", mc.groupID)
+		}
+	}
+}
+
+func (mc *MConsumerGroup) Close() error {
+	return mc.ConsumerGroup.Close()
+}
diff --git a/pkg/common/storage/kafka/producer.go b/pkg/common/storage/kafka/producer.go
@@ -0,0 +1,82 @@
+// Copyright © 2023 OpenIM. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kafka
+
+import (
+	"context"
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/tools/errs"
+	"google.golang.org/protobuf/proto"
+)
+
+// Producer represents a Kafka producer.
+type Producer struct {
+	addr     []string
+	topic    string
+	config   *sarama.Config
+	producer sarama.SyncProducer
+}
+
+func NewKafkaProducer(config *sarama.Config, addr []string, topic string) (*Producer, error) {
+	producer, err := NewProducer(config, addr)
+	if err != nil {
+		return nil, err
+	}
+	return &Producer{
+		addr:     addr,
+		topic:    topic,
+		config:   config,
+		producer: producer,
+	}, nil
+}
+
+// SendMessage sends a message to the Kafka topic configured in the Producer.
+func (p *Producer) SendMessage(ctx context.Context, key string, msg proto.Message) (int32, int64, error) {
+	// Marshal the protobuf message
+	bMsg, err := proto.Marshal(msg)
+	if err != nil {
+		return 0, 0, errs.WrapMsg(err, "kafka proto Marshal err")
+	}
+	if len(bMsg) == 0 {
+		return 0, 0, errs.WrapMsg(errEmptyMsg, "kafka proto Marshal err")
+	}
+
+	// Prepare Kafka message
+	kMsg := &sarama.ProducerMessage{
+		Topic: p.topic,
+		Key:   sarama.StringEncoder(key),
+		Value: sarama.ByteEncoder(bMsg),
+	}
+
+	// Validate message key and value
+	if kMsg.Key.Length() == 0 || kMsg.Value.Length() == 0 {
+		return 0, 0, errs.Wrap(errEmptyMsg)
+	}
+
+	// Attach context metadata as headers
+	header, err := GetMQHeaderWithContext(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	kMsg.Headers = header
+
+	// Send the message
+	partition, offset, err := p.producer.SendMessage(kMsg)
+	if err != nil {
+		return 0, 0, errs.WrapMsg(err, "p.producer.SendMessage error")
+	}
+
+	return partition, offset, nil
+}
diff --git a/pkg/common/storage/kafka/sarama.go b/pkg/common/storage/kafka/sarama.go
@@ -0,0 +1,85 @@
+package kafka
+
+import (
+	"bytes"
+	"strings"
+
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/tools/errs"
+)
+
+func BuildConsumerGroupConfig(conf *Config, initial int64, autoCommitEnable bool) (*sarama.Config, error) {
+	kfk := sarama.NewConfig()
+	kfk.Version = sarama.V2_0_0_0
+	kfk.Consumer.Offsets.Initial = initial
+	kfk.Consumer.Offsets.AutoCommit.Enable = autoCommitEnable
+	kfk.Consumer.Return.Errors = false
+	if conf.Username != "" || conf.Password != "" {
+		kfk.Net.SASL.Enable = true
+		kfk.Net.SASL.User = conf.Username
+		kfk.Net.SASL.Password = conf.Password
+	}
+	if conf.TLS.EnableTLS {
+		tls, err := newTLSConfig(conf.TLS.ClientCrt, conf.TLS.ClientKey, conf.TLS.CACrt, []byte(conf.TLS.ClientKeyPwd), conf.TLS.InsecureSkipVerify)
+		if err != nil {
+			return nil, err
+		}
+		kfk.Net.TLS.Config = tls
+		kfk.Net.TLS.Enable = true
+	}
+	return kfk, nil
+}
+
+func NewConsumerGroup(conf *sarama.Config, addr []string, groupID string) (sarama.ConsumerGroup, error) {
+	cg, err := sarama.NewConsumerGroup(addr, groupID, conf)
+	if err != nil {
+		return nil, errs.WrapMsg(err, "NewConsumerGroup failed", "addr", addr, "groupID", groupID, "conf", *conf)
+	}
+	return cg, nil
+}
+
+func BuildProducerConfig(conf Config) (*sarama.Config, error) {
+	kfk := sarama.NewConfig()
+	kfk.Producer.Return.Successes = true
+	kfk.Producer.Return.Errors = true
+	kfk.Producer.Partitioner = sarama.NewHashPartitioner
+	if conf.Username != "" || conf.Password != "" {
+		kfk.Net.SASL.Enable = true
+		kfk.Net.SASL.User = conf.Username
+		kfk.Net.SASL.Password = conf.Password
+	}
+	switch strings.ToLower(conf.ProducerAck) {
+	case "no_response":
+		kfk.Producer.RequiredAcks = sarama.NoResponse
+	case "wait_for_local":
+		kfk.Producer.RequiredAcks = sarama.WaitForLocal
+	case "wait_for_all":
+		kfk.Producer.RequiredAcks = sarama.WaitForAll
+	default:
+		kfk.Producer.RequiredAcks = sarama.WaitForAll
+	}
+	if conf.CompressType == "" {
+		kfk.Producer.Compression = sarama.CompressionNone
+	} else {
+		if err := kfk.Producer.Compression.UnmarshalText(bytes.ToLower([]byte(conf.CompressType))); err != nil {
+			return nil, errs.WrapMsg(err, "UnmarshalText failed", "compressType", conf.CompressType)
+		}
+	}
+	if conf.TLS.EnableTLS {
+		tls, err := newTLSConfig(conf.TLS.ClientCrt, conf.TLS.ClientKey, conf.TLS.CACrt, []byte(conf.TLS.ClientKeyPwd), conf.TLS.InsecureSkipVerify)
+		if err != nil {
+			return nil, err
+		}
+		kfk.Net.TLS.Config = tls
+		kfk.Net.TLS.Enable = true
+	}
+	return kfk, nil
+}
+
+func NewProducer(conf *sarama.Config, addr []string) (sarama.SyncProducer, error) {
+	producer, err := sarama.NewSyncProducer(addr, conf)
+	if err != nil {
+		return nil, errs.WrapMsg(err, "NewSyncProducer failed", "addr", addr, "conf", *conf)
+	}
+	return producer, nil
+}
diff --git a/pkg/common/storage/kafka/tls.go b/pkg/common/storage/kafka/tls.go
diff --git a/pkg/common/storage/kafka/util.go b/pkg/common/storage/kafka/util.go
diff --git a/pkg/common/storage/kafka/verify.go b/pkg/common/storage/kafka/verify.go
diff --git a/tools/seq/internal/seq.go b/tools/seq/internal/seq.go
diff --git a/tools/stress-test-v2/main.go b/tools/stress-test-v2/main.go

Original file line number	Diff line number	Diff line change
`@@ -8,7 +8,6 @@ import (`
`8`	`8`	`"github.com/gin-contrib/gzip"`
`9`	`9`	`"github.com/gin-gonic/gin"`
`10`	`10`	`"github.com/gin-gonic/gin/binding"`
`11`		`- "github.com/go-playground/validator/v10"`
`12`	`11`	`"github.com/openimsdk/open-im-server/v3/internal/api/jssdk"`
`13`	`12`	`"github.com/openimsdk/open-im-server/v3/pkg/authverify"`
`14`	`13`	`"github.com/openimsdk/open-im-server/v3/pkg/common/config"`
`@@ -290,6 +289,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf`
`290`	`289`	`conversationGroup.POST("/get_not_notify_conversation_ids", c.GetNotNotifyConversationIDs)`
`291`	`290`	`conversationGroup.POST("/get_pinned_conversation_ids", c.GetPinnedConversationIDs)`
`292`	`291`	`conversationGroup.POST("/delete_conversations", c.DeleteConversations)`
	`292`	`+ conversationGroup.POST("/update_conversations_by_user", c.UpdateConversationsByUser)`
`293`	`293`	`}`
`294`	`294`
`295`	`295`	`{`
Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ func (c *conversationServer) GetFullOwnerConversationIDs(ctx context.Context, re`
`27`	`27`	`conversationIDs = nil`
`28`	`28`	`}`
`29`	`29`	`return &conversation.GetFullOwnerConversationIDsResp{`
`30`		`- Version: idHash,`
	`30`	`+ Version: uint64(vl.Version),`
`31`	`31`	`VersionID: vl.ID.Hex(),`
`32`	`32`	`Equal: req.IdHash == idHash,`
`33`	`33`	`ConversationIDs: conversationIDs,`
Original file line number	Diff line number	Diff line change
`@@ -56,7 +56,7 @@ func (s friendServer) GetFullFriendUserIDs(ctx context.Context, req relation.G`
`56`	`56`	`userIDs = nil`
`57`	`57`	`}`
`58`	`58`	`return &relation.GetFullFriendUserIDsResp{`
`59`		`- Version: idHash,`
	`59`	`+ Version: uint64(vl.Version),`
`60`	`60`	`VersionID: vl.ID.Hex(),`
`61`	`61`	`Equal: req.IdHash == idHash,`
`62`	`62`	`UserIDs: userIDs,`
Original file line number	Diff line number	Diff line change
`@@ -13,7 +13,6 @@ import (`
`13`	`13`	`"github.com/openimsdk/tools/errs"`
`14`	`14`	`"github.com/openimsdk/tools/log"`
`15`	`15`	`"github.com/openimsdk/tools/utils/datautil"`
`16`		`- "github.com/redis/go-redis/v9"`
`17`	`16`	`)`
`18`	`17`
`19`	`18`	`type tokenCache struct {`
`@@ -220,7 +219,6 @@ func (c *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, p`
`220`	`219`	`if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil {`
`221`	`220`	`return errs.Wrap(err)`
`222`	`221`	`}`
`223`		`-`
`224`	`222`	`if c.localCache != nil {`
`225`	`223`	`c.removeLocalTokenCache(ctx, key)`
`226`	`224`	`}`