1use serde_json::{json, Map, Number, Value};
13
14use crate::local::filter;
15use crate::local::value::{self, get_path, remove_path, set_path};
16
17pub fn insert_one(rows: &mut Vec<Value>, doc: &Value) -> Result<Value, String> {
25 let doc = normalize_doc(doc)?;
26 if let Some(id) = doc.get("_id") {
27 if rows.iter().any(|r| id_equal(r, id)) {
28 return Err(format!("insertOne: _id 冲突({id})"));
29 }
30 }
31 rows.push(doc.clone());
32 Ok(doc)
33}
34
35pub fn insert_many(
40 rows: &mut Vec<Value>,
41 docs: &[Value],
42 upsert_by_id: bool,
43) -> Result<Value, String> {
44 let normalized = docs
45 .iter()
46 .map(normalize_doc)
47 .collect::<Result<Vec<_>, _>>()?;
48
49 if upsert_by_id {
50 for doc in normalized {
51 match doc.get("_id") {
52 Some(id) => match rows.iter_mut().find(|r| id_equal(r, id)) {
53 Some(slot) => *slot = doc,
54 None => rows.push(doc),
55 },
56 None => rows.push(doc),
57 }
58 }
59 return Ok(json!({ "insertedCount": docs.len() }));
60 }
61
62 for (i, doc) in normalized.iter().enumerate() {
63 if let Some(id) = doc.get("_id") {
64 if rows.iter().any(|r| id_equal(r, id)) {
65 return Err(format!("insertMany: _id 冲突({id})"));
66 }
67 if normalized[..i].iter().any(|prev| id_equal(prev, id)) {
68 return Err(format!("insertMany: 批内 _id 重复({id})"));
69 }
70 }
71 }
72 rows.extend(normalized);
73 Ok(json!({ "insertedCount": docs.len() }))
74}
75
76pub fn update_many(rows: &mut [Value], condition: &Value, update: &Value) -> Result<Value, String> {
84 let mut modified = 0_usize;
85 for row in rows.iter_mut() {
86 if filter::matches(row, condition)? {
87 let before = row.clone();
88 apply_update(row, update, false)?;
89 if !value::values_equal(&before, row) {
90 modified += 1;
91 }
92 }
93 }
94 Ok(json!({ "modifiedCount": modified }))
95}
96
97pub fn find_one_and_update(
102 rows: &mut Vec<Value>,
103 condition: &Value,
104 update: &Value,
105 options: &Value,
106) -> Result<Value, String> {
107 let upsert = options
108 .get("upsert")
109 .and_then(Value::as_bool)
110 .unwrap_or(false);
111 let return_before = options
112 .get("returnDocument")
113 .and_then(Value::as_str)
114 .map(|s| s == "before")
115 .unwrap_or(false);
116
117 if let Some(idx) = find_first(rows, condition)? {
118 let before = rows[idx].clone();
119 let mut doc = before.clone();
120 apply_update(&mut doc, update, false)?;
121 rows[idx] = doc.clone();
122 return Ok(if return_before { before } else { doc });
123 }
124 if !upsert {
125 return Ok(Value::Null);
126 }
127 let mut doc = build_upsert_base(condition);
128 apply_update(&mut doc, update, true)?;
129 rows.push(doc.clone());
130 Ok(if return_before { Value::Null } else { doc })
131}
132
133pub fn delete_many(rows: &mut Vec<Value>, condition: &Value) -> Result<Value, String> {
135 let mut kept = Vec::with_capacity(rows.len());
136 let mut deleted = 0_usize;
137 for row in rows.drain(..) {
138 if filter::matches(&row, condition)? {
139 deleted += 1;
140 } else {
141 kept.push(row);
142 }
143 }
144 *rows = kept;
145 Ok(json!({ "deletedCount": deleted }))
146}
147
148fn apply_update(doc: &mut Value, update: &Value, is_insert: bool) -> Result<(), String> {
154 let obj = update
155 .as_object()
156 .ok_or_else(|| format!("update 必须是对象,收到 {}", type_name(update)))?;
157 if obj.is_empty() {
158 return Err("没有提供要更新的字段".to_string());
159 }
160 if !doc.is_object() {
161 *doc = Value::Object(Map::new());
162 }
163 for (op, arg) in obj {
164 match op.as_str() {
165 "$set" => apply_set(doc, arg)?,
166 "$inc" => apply_inc(doc, arg)?,
167 "$unset" => apply_unset(doc, arg)?,
168 "$setOnInsert" => {
170 if is_insert {
171 apply_set(doc, arg)?;
172 }
173 }
174 other => {
175 return Err(format!(
176 "不支持的更新操作符: {other}(本地求值器仅支持 $set/$inc/$unset/$setOnInsert,拒绝静默)"
177 ));
178 }
179 }
180 }
181 Ok(())
182}
183
184fn apply_set(doc: &mut Value, arg: &Value) -> Result<(), String> {
186 let fields = arg
187 .as_object()
188 .ok_or_else(|| format!("$set 需要对象,收到 {}", type_name(arg)))?;
189 for (path, v) in fields {
190 set_path(doc, path, v.clone());
191 }
192 Ok(())
193}
194
195fn apply_inc(doc: &mut Value, arg: &Value) -> Result<(), String> {
197 let fields = arg
198 .as_object()
199 .ok_or_else(|| format!("$inc 需要对象,收到 {}", type_name(arg)))?;
200 for (path, inc) in fields {
201 if inc.as_f64().is_none() {
202 return Err(format!(
203 "$inc 的增量必须是数字,字段 \"{path}\" 收到 {}",
204 type_name(inc)
205 ));
206 }
207 let new = match get_path(doc, path) {
208 None => inc.clone(),
210 Some(cur) => {
211 let base = cur.as_f64().ok_or_else(|| {
212 format!(
213 "$inc 目标字段 \"{path}\" 非数值({}):Mongo 语义要求对数值字段自增",
214 type_name(cur)
215 )
216 })?;
217 let inc_f = inc.as_f64().unwrap_or(0.0);
218 match (cur.as_i64(), inc.as_i64()) {
219 (Some(a), Some(b)) => match a.checked_add(b) {
220 Some(s) => Value::from(s),
221 None => float_value(a as f64 + b as f64),
222 },
223 _ => float_value(base + inc_f),
224 }
225 }
226 };
227 set_path(doc, path, new);
228 }
229 Ok(())
230}
231
232fn apply_unset(doc: &mut Value, arg: &Value) -> Result<(), String> {
234 let fields = arg
235 .as_object()
236 .ok_or_else(|| format!("$unset 需要对象,收到 {}", type_name(arg)))?;
237 for path in fields.keys() {
238 remove_path(doc, path);
239 }
240 Ok(())
241}
242
243fn normalize_doc(v: &Value) -> Result<Value, String> {
249 match v {
250 Value::Object(_) => Ok(v.clone()),
251 Value::Null => Ok(Value::Object(Map::new())),
252 other => Err(format!(
253 "文档必须是对象,收到 {}(拒绝静默按空文档写入)",
254 type_name(other)
255 )),
256 }
257}
258
259fn id_equal(row: &Value, id: &Value) -> bool {
261 row.get("_id")
262 .map(|v| value::values_equal(v, id))
263 .unwrap_or(false)
264}
265
266fn find_first(rows: &[Value], condition: &Value) -> Result<Option<usize>, String> {
268 for (i, row) in rows.iter().enumerate() {
269 if filter::matches(row, condition)? {
270 return Ok(Some(i));
271 }
272 }
273 Ok(None)
274}
275
276fn build_upsert_base(condition: &Value) -> Value {
278 let mut m = Map::new();
279 if let Some(o) = condition.as_object() {
280 for (k, v) in o {
281 if k.starts_with('$') {
282 continue;
283 }
284 match v {
285 Value::Object(inner) => {
286 if let Some(eq) = inner.get("$eq") {
287 m.insert(k.clone(), eq.clone());
288 }
289 }
290 other => {
291 m.insert(k.clone(), other.clone());
292 }
293 }
294 }
295 }
296 Value::Object(m)
297}
298
299fn float_value(f: f64) -> Value {
300 Number::from_f64(f)
301 .map(Value::Number)
302 .unwrap_or(Value::Null)
303}
304
305fn type_name(v: &Value) -> &'static str {
306 match v {
307 Value::Null => "null",
308 Value::Bool(_) => "bool",
309 Value::Number(_) => "number",
310 Value::String(_) => "string",
311 Value::Array(_) => "array",
312 Value::Object(_) => "object",
313 }
314}
315
316#[cfg(test)]
317mod tests {
318 use super::*;
319 use serde_json::json;
320
321 fn rows(v: Value) -> Vec<Value> {
322 v.as_array().cloned().expect("文档数组")
323 }
324
325 #[test]
326 fn insert_one_appends_and_conflicts() {
327 let mut r = rows(json!([]));
328 let out = insert_one(&mut r, &json!({ "_id": "a", "n": 1 })).unwrap();
329 assert_eq!(out, json!({ "_id": "a", "n": 1 }));
330 assert_eq!(r.len(), 1);
331 assert!(insert_one(&mut r, &json!({ "_id": "a", "n": 2 })).is_err());
333 assert_eq!(r.len(), 1);
334 assert_eq!(insert_one(&mut r, &Value::Null).unwrap(), json!({}));
336 assert!(insert_one(&mut r, &json!(5)).is_err());
338 }
339
340 #[test]
341 fn insert_many_plain_and_conflicts() {
342 let mut r = rows(json!([{ "_id": "a" }]));
343 assert!(insert_many(&mut r, &[json!({ "_id": "a" })], false).is_err());
345 assert!(insert_many(
347 &mut r,
348 &[json!({ "_id": "b" }), json!({ "_id": "b" })],
349 false
350 )
351 .is_err());
352 let out = insert_many(
354 &mut r,
355 &[json!({ "_id": "b" }), json!({ "_id": "c" })],
356 false,
357 )
358 .unwrap();
359 assert_eq!(out, json!({ "insertedCount": 2 }));
360 assert_eq!(r.len(), 3);
361 }
362
363 #[test]
364 fn insert_many_upsert_by_id_is_idempotent() {
365 let mut r = rows(json!([{ "_id": "a", "n": 1 }]));
366 let out = insert_many(
368 &mut r,
369 &[json!({ "_id": "a", "n": 9 }), json!({ "_id": "z", "n": 5 })],
370 true,
371 )
372 .unwrap();
373 assert_eq!(out, json!({ "insertedCount": 2 }));
374 assert_eq!(
375 r,
376 rows(json!([{ "_id": "a", "n": 9 }, { "_id": "z", "n": 5 }]))
377 );
378 insert_many(&mut r, &[json!({ "_id": "a", "n": 9 })], true).unwrap();
380 assert_eq!(
381 r,
382 rows(json!([{ "_id": "a", "n": 9 }, { "_id": "z", "n": 5 }]))
383 );
384 }
385
386 #[test]
387 fn update_many_set_inc_unset() {
388 let mut r = rows(json!([
389 { "_id": 1, "s": "paid", "n": 1 },
390 { "_id": 2, "s": "open", "n": 2 }
391 ]));
392 let out = update_many(
393 &mut r,
394 &json!({ "s": "paid" }),
395 &json!({ "$set": { "tag": "x" }, "$inc": { "n": 10 }, "$unset": { "s": "" } }),
396 )
397 .unwrap();
398 assert_eq!(out, json!({ "modifiedCount": 1 }));
399 assert_eq!(r[0], json!({ "_id": 1, "n": 11, "tag": "x" }));
400 assert_eq!(r[1], json!({ "_id": 2, "s": "open", "n": 2 }));
401 let out2 = update_many(
403 &mut r,
404 &json!({ "_id": 1 }),
405 &json!({ "$set": { "n": 11 } }),
406 )
407 .unwrap();
408 assert_eq!(out2, json!({ "modifiedCount": 0 }));
409 }
410
411 #[test]
412 fn update_inc_type_and_non_numeric_error() {
413 let mut r = rows(json!([{ "_id": 1, "i": 2 }, { "_id": 2, "s": "x" }]));
414 update_many(&mut r, &json!({ "_id": 1 }), &json!({ "$inc": { "i": 3 } })).unwrap();
416 assert_eq!(r[0]["i"], json!(5));
417 update_many(
419 &mut r,
420 &json!({ "_id": 1 }),
421 &json!({ "$inc": { "gone": 1 } }),
422 )
423 .unwrap();
424 assert_eq!(r[0]["gone"], json!(1));
425 assert!(update_many(&mut r, &json!({ "_id": 2 }), &json!({ "$inc": { "s": 1 } })).is_err());
427 }
428
429 #[test]
430 fn update_unknown_operator_errors() {
431 let mut r = rows(json!([{ "_id": 1 }]));
432 assert!(update_many(&mut r, &json!({}), &json!({ "$push": { "a": 1 } })).is_err());
433 assert!(update_many(&mut r, &json!({}), &json!({})).is_err());
434 }
435
436 #[test]
437 fn find_one_and_update_after_before_and_upsert() {
438 let mut r = rows(json!([{ "_id": 1, "n": 1 }, { "_id": 2, "n": 2 }]));
439 let out = find_one_and_update(
441 &mut r,
442 &json!({ "_id": 1 }),
443 &json!({ "$set": { "n": 7 } }),
444 &json!({}),
445 )
446 .unwrap();
447 assert_eq!(out, json!({ "_id": 1, "n": 7 }));
448 let out2 = find_one_and_update(
450 &mut r,
451 &json!({ "_id": 1 }),
452 &json!({ "$set": { "n": 8 } }),
453 &json!({ "returnDocument": "before" }),
454 )
455 .unwrap();
456 assert_eq!(out2, json!({ "_id": 1, "n": 7 }));
457 assert_eq!(r[0], json!({ "_id": 1, "n": 8 }));
458 let miss = find_one_and_update(
460 &mut r,
461 &json!({ "_id": 99 }),
462 &json!({ "$set": { "n": 1 } }),
463 &json!({}),
464 )
465 .unwrap();
466 assert_eq!(miss, Value::Null);
467 assert_eq!(r.len(), 2);
468 let up = find_one_and_update(
470 &mut r,
471 &json!({ "_id": 3 }),
472 &json!({ "$set": { "n": 3 }, "$setOnInsert": { "_id": 3, "createdAt": 100 } }),
473 &json!({ "upsert": true }),
474 )
475 .unwrap();
476 assert_eq!(up, json!({ "_id": 3, "n": 3, "createdAt": 100 }));
477 assert_eq!(r.len(), 3);
478 }
479
480 #[test]
481 fn delete_many_counts() {
482 let mut r = rows(json!([
483 { "_id": 1, "s": "a" },
484 { "_id": 2, "s": "b" },
485 { "_id": 3, "s": "a" }
486 ]));
487 let out = delete_many(&mut r, &json!({ "s": "a" })).unwrap();
488 assert_eq!(out, json!({ "deletedCount": 2 }));
489 assert_eq!(r, rows(json!([{ "_id": 2, "s": "b" }])));
490 assert_eq!(
492 delete_many(&mut r, &json!({ "s": "nope" })).unwrap(),
493 json!({ "deletedCount": 0 })
494 );
495 }
496}