Merge 76ad8c88ac4f5936a172b55d933682a8d1c9fcb6 into a5ac7f2a8117ec6b283c74a9d6a4db578a2bdd37

internal/rpc/conversation/conversation.go

@@ -31,6 +31,7 @@ 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"
@@ -243,14 +244,11 @@ 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
 	}
@@ -302,7 +300,6 @@ 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 {
@@ -368,8 +365,6 @@ 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
@@ -377,7 +372,6 @@ 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
 		}
@@ -393,7 +387,6 @@ 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
 		}
@@ -409,7 +402,6 @@ 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
 		}
@@ -431,7 +423,6 @@ 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
@@ -717,7 +708,7 @@ func (c *conversationServer) GetConversationsNeedClearMsg(ctx context.Context, _
 
 	maxPage := (num + batchNum - 1) / batchNum
 
-	temp := make([]*dbModel.Conversation, 0, maxPage*batchNum)
+	temp := make([]*model.Conversation, 0, maxPage*batchNum)
 
 	for pageNumber := 0; pageNumber < int(maxPage); pageNumber++ {
 		pagination := &sdkws.RequestPagination{

internal/tools/cron/dist_look.go

@@ -52,12 +52,15 @@ func (e *EtcdLocker) ExecuteWithLock(ctx context.Context, taskName string, task
 
 	err = mutex.TryLock(ctxWithTimeout)
 	if err != nil {
-		// errors.Is(err, concurrency.ErrLocked)
-		log.ZDebug(ctx, "Task is being executed by another instance, skipping",
-			"taskName", taskName,
-			"instanceID", e.instanceID,
-			"error", err.Error())
-
+		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)
+		}
 		return
 	}