fix: 修复编译错误和 HTTP upsert 功能

- examples/stream_aggregate_example.go: 使用循环调用 Insert 替代不存在的 InsertMany
- internal/engine/crud_handler.go: Update 方法添加 upsert 参数
- internal/protocol/http/server.go: 传递 op.Upsert 到 CRUDHandler
- internal/engine/memory_store.go: upsert 时优先使用 filter 中的 _id

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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) (*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, _, err := h.store.Update(collection, filter, update, false, nil)
+	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

@@ -271,7 +271,7 @@ func (h *RequestHandler) HandleUpdate(w http.ResponseWriter, r *http.Request, db
 	upserted := make([]types.UpsertID, 0)
 
 	for _, op := range req.Updates {
-		result, err := h.crud.Update(context.Background(), fullCollection, op.Q, op.U)
+		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