From cd6168be3634a900e6c55ef7acc6e5c5d55e73f2 Mon Sep 17 00:00:00 2001
From: Kagari22 <fengweichen311@gmail.com>
Date: Tue, 18 Aug 2026 12:05:48 +0800
Subject: [PATCH] fix(conversation): tolerate concurrent duplicate creation

---
 .../storage/database/mgo/conversation.go      | 10 ++-
 .../storage/database/mgo/conversation_test.go | 76 +++++++++++++++++++
 2 files changed, 85 insertions(+), 1 deletion(-)
 create mode 100644 pkg/common/storage/database/mgo/conversation_test.go

diff --git a/pkg/common/storage/database/mgo/conversation.go b/pkg/common/storage/database/mgo/conversation.go
index 30f08f5e5..e197c7c87 100644
--- a/pkg/common/storage/database/mgo/conversation.go
+++ b/pkg/common/storage/database/mgo/conversation.go
@@ -71,7 +71,15 @@ type ConversationMgo struct {
 
 func (c *ConversationMgo) Create(ctx context.Context, conversations []*model.Conversation) (err error) {
 	return mongoutil.IncrVersion(func() error {
-		return mongoutil.InsertMany(ctx, c.coll, conversations)
+		err := mongoutil.InsertMany(ctx, c.coll, conversations, options.InsertMany().SetOrdered(false))
+		// Conversation creation is a derived, idempotent operation. A concurrent
+		// request may create the same (owner_user_id, conversation_id) document
+		// between the caller's read and this insert. In that case MongoDB reports a
+		// duplicate-key error even though the desired state already exists.
+		if mongo.IsDuplicateKeyError(err) {
+			return nil
+		}
+		return err
 	}, func() error {
 		userConversation := make(map[string][]string)
 		for _, conversation := range conversations {
diff --git a/pkg/common/storage/database/mgo/conversation_test.go b/pkg/common/storage/database/mgo/conversation_test.go
new file mode 100644
index 000000000..d06aae1a4
--- /dev/null
+++ b/pkg/common/storage/database/mgo/conversation_test.go
@@ -0,0 +1,76 @@
+package mgo
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo/integration/mtest"
+)
+
+type conversationVersionLogStub struct{}
+
+func (conversationVersionLogStub) IncrVersion(context.Context, string, []string, int32) error {
+	return nil
+}
+
+func (conversationVersionLogStub) FindChangeLog(context.Context, string, uint, int) (*model.VersionLog, error) {
+	return nil, nil
+}
+
+func (conversationVersionLogStub) BatchFindChangeLog(context.Context, []string, []uint, []int) ([]*model.VersionLog, error) {
+	return nil, nil
+}
+
+func (conversationVersionLogStub) DeleteAfterUnchangedLog(context.Context, time.Time) error {
+	return nil
+}
+
+func (conversationVersionLogStub) Delete(context.Context, string) error {
+	return nil
+}
+
+func TestConversationMgoCreateIgnoresDuplicateKey(t *testing.T) {
+	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
+
+	mt.Run("duplicate conversation is idempotent", func(mt *mtest.T) {
+		mt.AddMockResponses(bson.D{
+			{Key: "ok", Value: 1},
+			{Key: "writeErrors", Value: bson.A{
+				bson.D{{Key: "index", Value: 0}, {Key: "code", Value: 11000}, {Key: "errmsg", Value: "duplicate key"}},
+			}},
+		})
+
+		conversationDB := &ConversationMgo{coll: mt.Coll, version: conversationVersionLogStub{}}
+		err := conversationDB.Create(context.Background(), []*model.Conversation{{
+			OwnerUserID:    "user-1",
+			ConversationID: "group-1",
+		}})
+		if err != nil {
+			mt.Fatalf("expected duplicate conversation creation to be idempotent, got %v", err)
+		}
+	})
+}
+
+func TestConversationMgoCreatePropagatesNonDuplicateError(t *testing.T) {
+	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
+
+	mt.Run("non-duplicate insert errors are returned", func(mt *mtest.T) {
+		mt.AddMockResponses(bson.D{
+			{Key: "ok", Value: 0},
+			{Key: "code", Value: 13},
+			{Key: "errmsg", Value: "permission denied"},
+		})
+
+		conversationDB := &ConversationMgo{coll: mt.Coll, version: conversationVersionLogStub{}}
+		err := conversationDB.Create(context.Background(), []*model.Conversation{{
+			OwnerUserID:    "user-1",
+			ConversationID: "group-1",
+		}})
+		if err == nil {
+			mt.Fatal("expected non-duplicate insert error to be returned")
+		}
+	})
+}