Merge branch 'main' of ssh://git.kingecg.top:2222/kingecg/gomog

examples/stream_aggregate_example.go

@@ -36,9 +36,11 @@ func main() {
 		docs = append(docs, doc)
 	}
 
-	if err := store.InsertMany(collection, docs); err != nil {
+	for _, doc := range docs {
-		log.Printf("Error inserting documents: %v", err)
+		if err := store.Insert(collection, doc); err != nil {
-		return
+			log.Printf("Error inserting document: %v", err)
+			return
+		}
 	}
 
 	// 定义聚合管道

internal/engine/crud_handler.go

@@ -54,8 +54,8 @@ func (h *CRUDHandler) Insert(ctx context.Context, collection string, docs []map[
 }
 
 // Update 更新文档
-func (h *CRUDHandler) Update(ctx context.Context, collection string, filter types.Filter, update types.Update, upsert bool, arrayFilters []types.Filter) (*types.UpdateResult, error) {
+func (h *CRUDHandler) Update(ctx context.Context, collection string, filter types.Filter, update types.Update, upsert bool) (*types.UpdateResult, error) {
-	matched, modified, upsertedIDs, err := h.store.Update(collection, filter, update, upsert, arrayFilters)
+	matched, modified, _, err := h.store.Update(collection, filter, update, upsert, nil)
 	if err != nil {
 		return nil, err
 	}

internal/engine/memory_store.go

@@ -388,6 +388,12 @@ func (ms *MemoryStore) Update(collection string, filter types.Filter, update typ
 	if matched == 0 && upsert {
 		// 创建新文档
 		newID := generateID()
+		// 优先使用 filter 中的 _id
+		if idVal, ok := filter["_id"]; ok {
+			if idStr, ok := idVal.(string); ok && idStr != "" {
+				newID = idStr
+			}
+		}
 		newDoc := make(map[string]interface{})
 
 		// 应用更新（$setOnInsert 会生效）

internal/protocol/http/server.go

@@ -270,8 +270,8 @@ func (h *RequestHandler) HandleUpdate(w http.ResponseWriter, r *http.Request, db
 	totalModified := 0
 	upserted := make([]types.UpsertID, 0)
 
-	for idx, op := range req.Updates {
+	for _, op := range req.Updates {
-		result, err := h.crud.Update(context.Background(), fullCollection, op.Q, op.U, op.Upsert, op.ArrayFilters)
+		result, err := h.crud.Update(context.Background(), fullCollection, op.Q, op.U, op.Upsert)
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusInternalServerError)
 			return
@@ -281,9 +281,9 @@ func (h *RequestHandler) HandleUpdate(w http.ResponseWriter, r *http.Request, db
 		totalModified += result.NModified
 
 		// 合并 upserted IDs
-		for _, uid := range result.Upserted {
+		for i, uid := range result.Upserted {
 			upserted = append(upserted, types.UpsertID{
-				Index: idx + uid.Index,
+				Index: len(upserted) + i,
 				ID:    uid.ID,
 			})
 		}