Merge e2c1bb0eeea43a036d0af05c38313ad01919767d into 5e813ef07986bfef8ee33a4d9f5f95d015a58774

config/mongodb.yml

@@ -1,7 +1,7 @@
 # URI for database connection, leave empty if using address and credential settings directly
-uri: 
+uri:
 # List of MongoDB server addresses
-address: [ localhost:37017 ]
+address: [localhost:37017]
 # Name of the database
 database: openim_v3
 # Username for database authentication
@@ -14,3 +14,38 @@ authSource: openim_v3
 maxPoolSize: 100
 # Maximum number of retry attempts for a failed database connection
 maxRetry: 10
+# MongoDB Mode, including "standalone", "replicaSet"
+mongoMode: "standalone"
+
+# The following configurations only take effect when mongoMode is set to "replicaSet"
+replicaSet:
+  name: rs0
+  hosts: [127.0.0.1:37017, 127.0.0.1:37018, 127.0.0.1:37019]
+  # Read concern level: "local", "available", "majority", "linearizable", "snapshot"
+  readConcern: majority
+  # maximum staleness of data in seconds
+  maxStaleness: 90s
+
+# The following configurations only take effect when mongoMode is set to "replicaSet"
+readPreference:
+  # Read preference mode, can be "primary", "primaryPreferred", "secondary", "secondaryPreferred", "nearest"
+  mode: primary
+  maxStaleness: 90s
+  # TagSets is an array of maps with priority based on order, empty map must be placed last for fallback tagSets
+  tagSets:
+    - datacenter: "cn-east"
+      rack: "1"
+      storage: "ssd"
+    - datacenter: "cn-east"
+      storage: "ssd"
+    - datacenter: "cn-east"
+    - {} # Empty map, indicates any node
+
+# The following configurations only take effect when mongoMode is set to "replicaSet"
+writeConcern:
+  # Write node count or tag (int, "majority", or custom tag)
+  w: majority
+  # Whether to wait for journal confirmation
+  j: true
+  # Write timeout duration
+  wtimeout: 30s

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.12
-	github.com/openimsdk/tools v0.0.50-alpha.92
+	github.com/openimsdk/tools v0.0.50-alpha.96
 	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.12 h1:2NYawXeHChYUeSme6QJ9pOLh+Empce2WmwEtbP4JvKk=
 github.com/openimsdk/protocol v0.0.73-alpha.12/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
-github.com/openimsdk/tools v0.0.50-alpha.92 h1:hWfykMhmi7EQEiwgQccJqbgggIuhun/PrVkBnjmj9Ec=
-github.com/openimsdk/tools v0.0.50-alpha.92/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
+github.com/openimsdk/tools v0.0.50-alpha.96 h1:U44Fq2jHiEvGi9zuYAnTRNx3Xd9T7P/kBAZLHvQ8xg4=
+github.com/openimsdk/tools v0.0.50-alpha.96/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/rpc/conversation/conversation.go

@@ -31,7 +31,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	dbModel "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache"
@@ -244,11 +243,14 @@ func (c *conversationServer) getConversations(ctx context.Context, ownerUserID s
 	return convert.ConversationsDB2Pb(conversations), nil
 }
 
+// Deprecated
 func (c *conversationServer) SetConversation(ctx context.Context, req *pbconversation.SetConversationReq) (*pbconversation.SetConversationResp, error) {
 	if err := authverify.CheckAccess(ctx, req.GetConversation().GetUserID()); err != nil {
 		return nil, err
 	}
 	var conversation dbModel.Conversation
+	conversation.CreateTime = time.Now()
+
 	if err := datautil.CopyStructFields(&conversation, req.Conversation); err != nil {
 		return nil, err
 	}
@@ -300,6 +302,7 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver
 	conversation.ConversationType = req.Conversation.ConversationType
 	conversation.UserID = req.Conversation.UserID
 	conversation.GroupID = req.Conversation.GroupID
+	conversation.CreateTime = time.Now()
 
 	m, conversation, err := UpdateConversationsMap(ctx, req)
 	if err != nil {
@@ -365,6 +368,8 @@ func (c *conversationServer) UpdateConversationsByUser(ctx context.Context, req
 // create conversation without notification for msg redis transfer.
 func (c *conversationServer) CreateSingleChatConversations(ctx context.Context, req *pbconversation.CreateSingleChatConversationsReq) (*pbconversation.CreateSingleChatConversationsResp, error) {
 	var conversation dbModel.Conversation
+	conversation.CreateTime = time.Now()
+
 	switch req.ConversationType {
 	case constant.SingleChatType:
 		// sendUser create
@@ -372,6 +377,7 @@ func (c *conversationServer) CreateSingleChatConversations(ctx context.Context,
 		conversation.ConversationType = req.ConversationType
 		conversation.OwnerUserID = req.SendID
 		conversation.UserID = req.RecvID
+
 		if err := c.webhookBeforeCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateSingleChatConversations, &conversation); err != nil && err != servererrs.ErrCallbackContinue {
 			return nil, err
 		}
@@ -387,6 +393,7 @@ func (c *conversationServer) CreateSingleChatConversations(ctx context.Context,
 		conversation2 := conversation
 		conversation2.OwnerUserID = req.RecvID
 		conversation2.UserID = req.SendID
+
 		if err := c.webhookBeforeCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateSingleChatConversations, &conversation); err != nil && err != servererrs.ErrCallbackContinue {
 			return nil, err
 		}
@@ -402,6 +409,7 @@ func (c *conversationServer) CreateSingleChatConversations(ctx context.Context,
 		conversation.ConversationType = req.ConversationType
 		conversation.OwnerUserID = req.RecvID
 		conversation.UserID = req.SendID
+
 		if err := c.webhookBeforeCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateSingleChatConversations, &conversation); err != nil && err != servererrs.ErrCallbackContinue {
 			return nil, err
 		}
@@ -423,6 +431,7 @@ func (c *conversationServer) CreateGroupChatConversations(ctx context.Context, r
 	conversation.ConversationID = msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupID)
 	conversation.GroupID = req.GroupID
 	conversation.ConversationType = constant.ReadGroupChatType
+	conversation.CreateTime = time.Now()
 
 	if err := c.webhookBeforeCreateGroupChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateGroupChatConversations, &conversation); err != nil {
 		return nil, err
@@ -708,7 +717,7 @@ func (c *conversationServer) GetConversationsNeedClearMsg(ctx context.Context, _
 
 	maxPage := (num + batchNum - 1) / batchNum
 
-	temp := make([]*model.Conversation, 0, maxPage*batchNum)
+	temp := make([]*dbModel.Conversation, 0, maxPage*batchNum)
 
 	for pageNumber := 0; pageNumber < int(maxPage); pageNumber++ {
 		pagination := &sdkws.RequestPagination{

internal/tools/cron/dist_look.go

@@ -52,15 +52,12 @@ func (e *EtcdLocker) ExecuteWithLock(ctx context.Context, taskName string, task
 
 	err = mutex.TryLock(ctxWithTimeout)
 	if err != nil {
-		if err == context.DeadlineExceeded {
-			log.ZDebug(ctx, "Task is being executed by another instance, skipping",
-				"taskName", taskName,
-				"instanceID", e.instanceID)
-		} else {
-			log.ZWarn(ctx, "Failed to acquire task lock", err,
-				"taskName", taskName,
-				"instanceID", e.instanceID)
-		}
+		// errors.Is(err, concurrency.ErrLocked)
+		log.ZDebug(ctx, "Task is being executed by another instance, skipping",
+			"taskName", taskName,
+			"instanceID", e.instanceID,
+			"error", err.Error())
+
 		return
 	}
 

pkg/common/config/config.go

@@ -71,15 +71,39 @@ type Minio struct {
 }
 
 type Mongo struct {
-	URI         string   `yaml:"uri"`
-	Address     []string `yaml:"address"`
-	Database    string   `yaml:"database"`
-	Username    string   `yaml:"username"`
-	Password    string   `yaml:"password"`
-	AuthSource  string   `yaml:"authSource"`
-	MaxPoolSize int      `yaml:"maxPoolSize"`
-	MaxRetry    int      `yaml:"maxRetry"`
+	URI            string   `yaml:"uri"`
+	Address        []string `yaml:"address"`
+	Database       string   `yaml:"database"`
+	Username       string   `yaml:"username"`
+	Password       string   `yaml:"password"`
+	AuthSource     string   `yaml:"authSource"`
+	MaxPoolSize    int      `yaml:"maxPoolSize"`
+	MaxRetry       int      `yaml:"maxRetry"`
+	MongoMode      string   `yaml:"mongoMode"`
+	ReplicaSet     ReplicaSetConfig
+	ReadPreference ReadPrefConfig
+	WriteConcern   WriteConcernConfig
 }
+
+type ReplicaSetConfig struct {
+	Name         string        `yaml:"name"`
+	Hosts        []string      `yaml:"hosts"`
+	ReadConcern  string        `yaml:"readConcern"`
+	MaxStaleness time.Duration `yaml:"maxStaleness"`
+}
+
+type ReadPrefConfig struct {
+	Mode         string              `yaml:"mode"`
+	TagSets      []map[string]string `yaml:"tagSets"`
+	MaxStaleness time.Duration       `yaml:"maxStaleness"`
+}
+
+type WriteConcernConfig struct {
+	W        any           `yaml:"w"`
+	J        bool          `yaml:"j"`
+	WTimeout time.Duration `yaml:"wtimeout"`
+}
+
 type Kafka struct {
 	Username           string   `yaml:"username"`
 	Password           string   `yaml:"password"`
@@ -493,6 +517,23 @@ func (m *Mongo) Build() *mongoutil.Config {
 		AuthSource:  m.AuthSource,
 		MaxPoolSize: m.MaxPoolSize,
 		MaxRetry:    m.MaxRetry,
+		MongoMode:   m.MongoMode,
+		ReplicaSet: &mongoutil.ReplicaSetConfig{
+			Name:         m.ReplicaSet.Name,
+			Hosts:        m.ReplicaSet.Hosts,
+			ReadConcern:  m.ReplicaSet.ReadConcern,
+			MaxStaleness: m.ReplicaSet.MaxStaleness,
+		},
+		ReadPreference: &mongoutil.ReadPrefConfig{
+			Mode:         m.ReadPreference.Mode,
+			TagSets:      m.ReadPreference.TagSets,
+			MaxStaleness: m.ReadPreference.MaxStaleness,
+		},
+		WriteConcern: &mongoutil.WriteConcernConfig{
+			W:        m.WriteConcern.W,
+			J:        m.WriteConcern.J,
+			WTimeout: m.WriteConcern.WTimeout,
+		},
 	}
 }