chore(server): item import performance enhancement (#1319)

server/internal/infrastructure/memory/item.go

@@ -24,7 +24,13 @@ type Item struct {
 	err  error
 }
 
-func (r *Item) FindByAssets(ctx context.Context, list id.AssetIDList, ref *version.Ref) (item.VersionedList, error) {
+func NewItem() repo.Item {
+	return &Item{
+		data: memorygit.NewVersionedSyncMap[item.ID, *item.Item](),
+	}
+}
+
+func (r *Item) FindByAssets(_ context.Context, list id.AssetIDList, ref *version.Ref) (item.VersionedList, error) {
 	if r.err != nil {
 		return nil, r.err
 	}
@@ -41,12 +47,6 @@ func (r *Item) FindByAssets(ctx context.Context, list id.AssetIDList, ref *versi
 	return res, nil
 }
 
-func NewItem() repo.Item {
-	return &Item{
-		data: memorygit.NewVersionedSyncMap[item.ID, *item.Item](),
-	}
-}
-
 func (r *Item) Filtered(filter repo.ProjectFilter) repo.Item {
 	return &Item{
 		data: r.data,
@@ -109,7 +109,7 @@ func (r *Item) FindByIDs(_ context.Context, list id.ItemIDList, ref *version.Ref
 	return r.data.LoadAll(list, lo.ToPtr(ref.OrLatest().OrVersion())), nil
 }
 
-func (r *Item) FindVersionByID(ctx context.Context, itemID id.ItemID, ver version.VersionOrRef) (item.Versioned, error) {
+func (r *Item) FindVersionByID(_ context.Context, itemID id.ItemID, ver version.VersionOrRef) (item.Versioned, error) {
 	if r.err != nil {
 		return nil, r.err
 	}
@@ -133,7 +133,7 @@ func (r *Item) FindAllVersionsByID(_ context.Context, id id.ItemID) (item.Versio
 	}), nil
 }
 
-func (r *Item) FindAllVersionsByIDs(ctx context.Context, ids id.ItemIDList) (item.VersionedList, error) {
+func (r *Item) FindAllVersionsByIDs(_ context.Context, ids id.ItemIDList) (item.VersionedList, error) {
 	if r.err != nil {
 		return nil, r.err
 	}
@@ -145,7 +145,7 @@ func (r *Item) FindAllVersionsByIDs(ctx context.Context, ids id.ItemIDList) (ite
 	}), nil
 }
 
-func (r *Item) LastModifiedByModel(ctx context.Context, modelID id.ModelID) (time.Time, error) {
+func (r *Item) LastModifiedByModel(_ context.Context, modelID id.ModelID) (time.Time, error) {
 	if r.err != nil {
 		return time.Time{}, r.err
 	}
@@ -175,7 +175,21 @@ func (r *Item) Save(_ context.Context, t *item.Item) error {
 	return nil
 }
 
-func (r *Item) UpdateRef(ctx context.Context, item id.ItemID, ref version.Ref, vr *version.VersionOrRef) error {
+func (r *Item) SaveAll(_ context.Context, il item.List) error {
+	if r.err != nil {
+		return r.err
+	}
+
+	for _, t := range il {
+		if !r.f.CanWrite(t.Project()) {
+			return repo.ErrOperationDenied
+		}
+	}
+	r.data.SaveAll(il.IDs(), il, nil)
+	return nil
+}
+
+func (r *Item) UpdateRef(_ context.Context, item id.ItemID, ref version.Ref, vr *version.VersionOrRef) error {
 	if r.err != nil {
 		return r.err
 	}

server/internal/infrastructure/memory/memorygit/memory.go

@@ -75,6 +75,18 @@ func (m *VersionedSyncMap[K, V]) SaveOne(key K, value V, parent *version.Version
 	}
 }
 
+func (m *VersionedSyncMap[K, V]) SaveAll(key []K, value []V, parent []*version.VersionOrRef) {
+	if len(key) != len(value) || (parent != nil && len(key) != len(parent)) {
+		return
+	}
+	if len(key) == 0 {
+		return
+	}
+	for i := 0; i < len(key); i++ {
+		m.SaveOne(key[i], value[i], parent[i])
+	}
+}
+
 func (m *VersionedSyncMap[K, V]) UpdateRef(key K, ref version.Ref, vr *version.VersionOrRef) {
 	m.Range(func(k K, v *version.Values[V]) bool {
 		if k == key {

server/internal/infrastructure/memory/thread.go

@@ -35,6 +35,20 @@ func (r *Thread) Save(_ context.Context, th *thread.Thread) error {
 	return nil
 }
 
+func (r *Thread) SaveAll(_ context.Context, th thread.List) error {
+	if r.err != nil {
+		return r.err
+	}
+
+	for _, t := range th {
+		if !r.f.CanWrite(t.Workspace()) {
+			return repo.ErrOperationDenied
+		}
+	}
+	r.data.StoreAll(th.ToMap())
+	return nil
+}
+
 func (r *Thread) Filtered(f repo.WorkspaceFilter) repo.Thread {
 	return &Thread{
 		data: r.data,

server/internal/infrastructure/mongo/item.go

@@ -13,6 +13,7 @@ import (
 	"github.com/reearth/reearthx/mongox"
 	"github.com/reearth/reearthx/rerror"
 	"github.com/reearth/reearthx/usecasex"
+	"github.com/samber/lo"
 	"go.mongodb.org/mongo-driver/bson"
 )
 
@@ -228,6 +229,20 @@ func (r *Item) Save(ctx context.Context, item *item.Item) error {
 	return r.client.SaveOne(ctx, id, doc, nil)
 }
 
+func (r *Item) SaveAll(ctx context.Context, items item.List) error {
+	if len(items) == 0 {
+		return nil
+	}
+
+	for _, itm := range items {
+		if !r.f.CanWrite(itm.Project()) {
+			return repo.ErrOperationDenied
+		}
+	}
+	docs, ids := mongodoc.NewItems(items)
+	return r.client.SaveAll(ctx, ids, lo.ToAnySlice(docs), nil)
+}
+
 func (r *Item) UpdateRef(ctx context.Context, item id.ItemID, ref version.Ref, vr *version.VersionOrRef) error {
 	return r.client.UpdateRef(ctx, item.String(), ref, vr)
 }

server/internal/infrastructure/mongo/mongodoc/thread.go

@@ -40,6 +40,20 @@ func NewThread(a *thread.Thread) (*ThreadDocument, string) {
 	return thd, id
 }
 
+func NewThreads(a thread.List) ([]ThreadDocument, []string) {
+	res := make([]ThreadDocument, 0, len(a))
+	ids := make([]string, 0, len(a))
+	for _, th := range a {
+		if th == nil {
+			continue
+		}
+		thDoc, thId := NewThread(th)
+		res = append(res, *thDoc)
+		ids = append(ids, thId)
+	}
+	return res, ids
+}
+
 func (d *ThreadDocument) Model() (*thread.Thread, error) {
 	thid, err := id.ThreadIDFrom(d.ID)
 	if err != nil {

server/internal/infrastructure/mongo/mongogit/collection.go

@@ -105,6 +105,27 @@ func (c *Collection) SaveOne(ctx context.Context, id string, d any, parent *vers
 	return nil
 }
 
+func (c *Collection) SaveAll(ctx context.Context, ids []string, docs []any, parents []*version.VersionOrRef) error {
+	// TODO: optimize to use bulk write
+	if len(ids) != len(docs) || (parents != nil && len(ids) != len(parents)) {
+		return rerror.ErrInvalidParams
+	}
+	if len(ids) == 0 {
+		return nil
+	}
+	for i := 0; i < len(ids); i++ {
+		var parent *version.VersionOrRef = nil
+		if parents != nil {
+			parent = parents[i]
+		}
+		err := c.SaveOne(ctx, ids[i], docs[i], parent)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (c *Collection) UpdateRef(ctx context.Context, id string, ref version.Ref, dest *version.VersionOrRef) error {
 	if _, err := c.client.Client().UpdateMany(ctx, bson.M{
 		"id":    id,

server/internal/infrastructure/mongo/thread.go

@@ -9,6 +9,7 @@ import (
 	"github.com/reearth/reearth-cms/server/pkg/thread"
 	"github.com/reearth/reearthx/mongox"
 	"github.com/reearth/reearthx/rerror"
+	"github.com/samber/lo"
 	"go.mongodb.org/mongo-driver/bson"
 )
 
@@ -39,6 +40,19 @@ func (r *ThreadRepo) Save(ctx context.Context, thread *thread.Thread) error {
 	return r.client.SaveOne(ctx, id, doc)
 }
 
+func (r *ThreadRepo) SaveAll(ctx context.Context, threads thread.List) error {
+	if len(threads) == 0 {
+		return nil
+	}
+	for _, t := range threads {
+		if !r.f.CanWrite(t.Workspace()) {
+			return repo.ErrOperationDenied
+		}
+	}
+	docs, ids := mongodoc.NewThreads(threads)
+	return r.client.SaveAll(ctx, ids, lo.ToAnySlice(docs))
+}
+
 func (r *ThreadRepo) Filtered(f repo.WorkspaceFilter) repo.Thread {
 	return &ThreadRepo{
 		client: r.client,

server/internal/usecase/interactor/item.go

@@ -366,16 +366,41 @@ func (i Item) Import(ctx context.Context, param interfaces.ImportItemsParam, ope
 		}
 		res := NewImportRes()
 
+		prj, err := i.repos.Project.FindByID(ctx, s.Project())
+		if err != nil {
+			return interfaces.ImportItemsResponse{}, err
+		}
+
 		m, err := i.repos.Model.FindByID(ctx, param.ModelID)
 		if err != nil {
 			return interfaces.ImportItemsResponse{}, err
 		}
 
+		itemsIds := lo.FilterMap(param.Items, func(i interfaces.ImportItemParam, _ int) (item.ID, bool) {
+			if i.ItemId != nil {
+				return *i.ItemId, true
+			}
+			return item.ID{}, false
+		})
+		oldItems, err := i.repos.Item.FindByIDs(ctx, itemsIds, nil)
+		if err != nil {
+			return interfaces.ImportItemsResponse{}, err
+		}
+
 		isMetadata := false
 		if m.Metadata() != nil && s.ID() == *m.Metadata() {
 			isMetadata = true
 		}
 
+		threadsToSave := thread.List{}
+		itemsToSave := item.List{}
+
+		type itemChanges struct {
+			oldFields item.Fields
+			action    interfaces.ImportStrategyType
+		}
+		itemsEvent := map[item.ID]itemChanges{}
+
 		// update schema if needed
 		if param.MutateSchema && len(param.Fields) > 0 {
 			for _, fieldParam := range param.Fields {
@@ -410,10 +435,7 @@ func (i Item) Import(ctx context.Context, param interfaces.ImportItemsParam, ope
 
 			var oldItem *item.Item
 			if itemParam.ItemId != nil {
-				itm, err := i.repos.Item.FindByID(ctx, *itemParam.ItemId, nil)
-				if err != nil && !errors.Is(err, rerror.ErrNotFound) {
-					return interfaces.ImportItemsResponse{}, err
-				}
+				itm := oldItems.Item(*itemParam.ItemId)
 				oldItem = itm.Value()
 			}
 
@@ -454,9 +476,7 @@ func (i Item) Import(ctx context.Context, param interfaces.ImportItemsParam, ope
 				if err != nil {
 					return interfaces.ImportItemsResponse{}, err
 				}
-				if err := i.repos.Thread.Save(ctx, th); err != nil {
-					return interfaces.ImportItemsResponse{}, err
-				}
+				threadsToSave = append(threadsToSave, th)
 
 				ib := item.New().
 					NewID().
@@ -508,9 +528,7 @@ func (i Item) Import(ctx context.Context, param interfaces.ImportItemsParam, ope
 					return interfaces.ImportItemsResponse{}, interfaces.ErrMetadataMismatch
 				}
 				mi.Value().SetOriginalItem(it.ID())
-				if err := i.repos.Item.Save(ctx, mi.Value()); err != nil {
-					return interfaces.ImportItemsResponse{}, err
-				}
+				itemsToSave = append(itemsToSave, mi.Value())
 			}
 
 			modelSchemaFields, otherFields := filterFieldParamsBySchema(itemParam.Fields, s)
@@ -527,7 +545,7 @@ func (i Item) Import(ctx context.Context, param interfaces.ImportItemsParam, ope
 			oldFields := it.Fields()
 			it.UpdateFields(fields)
 
-			groupFields, groupSchemas, err := i.handleGroupFields(ctx, otherFields, s, m.ID(), it.Fields())
+			groupFields, _, err := i.handleGroupFields(ctx, otherFields, s, m.ID(), it.Fields())
 			if err != nil {
 				return interfaces.ImportItemsResponse{}, err
 			}
@@ -538,33 +556,42 @@ func (i Item) Import(ctx context.Context, param interfaces.ImportItemsParam, ope
 				return interfaces.ImportItemsResponse{}, err
 			}
 
-			if err := i.repos.Item.Save(ctx, it); err != nil {
-				return interfaces.ImportItemsResponse{}, err
-			}
+			itemsToSave = append(itemsToSave, it)
 
 			if isMetadata {
 				continue
 			}
-
-			vi, err := i.repos.Item.FindByID(ctx, it.ID(), nil)
-			if err != nil {
-				return interfaces.ImportItemsResponse{}, err
+			itemsEvent[it.ID()] = itemChanges{
+				oldFields: oldFields,
+				action:    action,
 			}
+		}
 
-			refItems, err := i.getReferencedItems(ctx, it.Fields())
-			if err != nil {
-				return interfaces.ImportItemsResponse{}, err
-			}
+		if err := i.repos.Thread.SaveAll(ctx, threadsToSave); err != nil {
+			return interfaces.ImportItemsResponse{}, err
+		}
 
-			prj, err := i.repos.Project.FindByID(ctx, s.Project())
+		if err := i.repos.Item.SaveAll(ctx, itemsToSave); err != nil {
+			return interfaces.ImportItemsResponse{}, err
+		}
+
+		//  TODO: create ItemsImported event
+		items, err := i.repos.Item.FindByIDs(ctx, lo.Keys(itemsEvent), nil)
+		if err != nil {
+			return interfaces.ImportItemsResponse{}, err
+		}
+
+		for k, changes := range itemsEvent {
+			vi := items.Item(k)
+			it := vi.Value()
+
+			refItems, err := i.getReferencedItems(ctx, it.Fields())
 			if err != nil {
 				return interfaces.ImportItemsResponse{}, err
 			}
 
-			// TODO: check if event creation is transactional
-			//  A: in future create ItemsImported event
 			var eType event.Type
-			if action == interfaces.ImportStrategyTypeInsert {
+			if changes.action == interfaces.ImportStrategyTypeInsert {
 				eType = event.ItemCreate
 				res.ItemInserted()
 			} else {
@@ -580,9 +607,9 @@ func (i Item) Import(ctx context.Context, param interfaces.ImportItemsParam, ope
 					Item:            vi.Value(),
 					Model:           m,
 					Schema:          s,
-					GroupSchemas:    groupSchemas,
+					GroupSchemas:    param.SP.GroupSchemas(),
 					ReferencedItems: refItems,
-					Changes:         item.CompareFields(it.Fields(), oldFields),
+					Changes:         item.CompareFields(it.Fields(), changes.oldFields),
 				},
 				Operator: operator.Operator(),
 			}); err != nil {
@@ -1021,6 +1048,7 @@ func (i Item) getItemCorrespondingItems(ctx context.Context, fr schema.FieldRefe
 }
 
 func (i Item) handleGroupFields(ctx context.Context, params []interfaces.ItemFieldParam, s *schema.Schema, mId id.ModelID, itemFields item.Fields) (item.Fields, schema.List, error) {
+	// TODO: use schema package to enhance performance
 	var res item.Fields
 	var groupSchemas schema.List
 	for _, field := range itemFields.FieldsByType(value.TypeGroup) {

server/internal/usecase/repo/Item.go

@@ -32,6 +32,7 @@ type Item interface {
 	FindByModelAndValue(context.Context, id.ModelID, []FieldAndValue, *version.Ref) (item.VersionedList, error)
 	IsArchived(context.Context, id.ItemID) (bool, error)
 	Save(context.Context, *item.Item) error
+	SaveAll(context.Context, item.List) error
 	UpdateRef(context.Context, id.ItemID, version.Ref, *version.VersionOrRef) error
 	Remove(context.Context, id.ItemID) error
 	Archive(context.Context, id.ItemID, id.ProjectID, bool) error