Skip to main content

rust_store_core/local/
update.rs

1//! 本地磁盘数据源 —— 写算子求值(`insertOne` / `insertMany` / `updateMany` /
2//! `findOneAndUpdate` / `deleteMany`)。
3//!
4//! 语义基准 = **MongoDB 驱动语义**(local 与 mongo 走同一条命令路径,必须同结果)。
5//! 纯逻辑:无 IO / 无时钟 / 无随机 —— 一切不确定性输入(`_id`、时间戳)均由 core /
6//! 宿主在命令中给足(core 不生成 `_id`)。
7//!
8//! 更新算子仅支持 `$set` / `$inc` / `$unset`(+ upsert 分支的 `$setOnInsert`,
9//! 对齐 `dialect/write/update.rs` 第 35–47 行);其余操作符(`$push` / `$addToSet` /
10//! `$pull` …)一律 `Err`(禁静默丢弃)。
11
12use serde_json::{json, Map, Number, Value};
13
14use crate::local::filter;
15use crate::local::value::{self, get_path, remove_path, set_path};
16
17// ---------------------------------------------------------------------------
18// insertOne / insertMany
19// ---------------------------------------------------------------------------
20
21/// `insertOne`:追加 `cmd.doc`;同 `_id` 冲突 → `Err`(对齐 Mongo 唯一键冲突)。
22///
23/// 返回写入后的文档(与驱动 `insertOne` 返回文档形态一致)。
24pub 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
35/// `insertMany`:`upsertById == true` → 按 `_id` **replace**(归档幂等,对齐
36/// `mongo.js` 第 136–143 行);否则同 `_id` 冲突(含批内重复)→ `Err`。
37///
38/// 返回 `{"insertedCount": n}`。
39pub 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
76// ---------------------------------------------------------------------------
77// updateMany / findOneAndUpdate / deleteMany
78// ---------------------------------------------------------------------------
79
80/// `updateMany`:过滤命中项全部应用 `cmd.update`;返回 `{"modifiedCount": n}`。
81///
82/// `modifiedCount` 为**实际发生变更**的文档数(对齐驱动语义:`$set` 同值不计)。
83pub 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
97/// `findOneAndUpdate`:首条命中应用更新;
98/// - 无命中且 `options.upsert == true` → 以过滤条件等值字段为底、应用 `$set`/`$setOnInsert` 后插入;
99/// - 无命中且非 upsert → `null`;
100/// - `returnDocument == "before"` → 更新前文档(upsert 首次插入时为 `null`,对齐驱动),否则返回更新后文档。
101pub 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
133/// `deleteMany`:删除命中项;返回 `{"deletedCount": n}`。
134pub 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
148// ---------------------------------------------------------------------------
149// 更新算子
150// ---------------------------------------------------------------------------
151
152/// 应用更新文档:逐算子求值(`$set` / `$inc` / `$unset`;`$setOnInsert` 仅插入分支生效)。
153fn 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            // upsert 专用:仅在插入分支生效,非插入路径忽略(对齐 dialect/write/update.rs 第 47 行)
169            "$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
184/// `$set`:点号路径逐字段写入(对齐 Mongo;含显式 `null` 亦写入)。
185fn 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
195/// `$inc`:数值自增;字段缺失视作 `0`(结果保留增量类型);目标非数值 → `Err`(对齐 Mongo)。
196fn 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            // 缺失 → 视为 0 + 增量 = 增量本身(保留整型/浮点形态)
209            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
232/// `$unset`:点号路径逐字段删除(取值无关,对齐 Mongo)。
233fn 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
243// ---------------------------------------------------------------------------
244// 工具
245// ---------------------------------------------------------------------------
246
247/// 文档规范化:对象原样;`null` → 空对象(对齐 `mongo.js` 的 `cmd.doc || {}`);其余 → `Err`。
248fn 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
259/// `_id` 等值判定(候选文档的 `_id` 与目标 `_id`)。
260fn 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
266/// 首条命中下标。
267fn 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
276/// upsert 新文档基底:取过滤条件中的**顶层等值字段**(`{k: v}` 或 `{k: {$eq: v}}`)。
277fn 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        // 同 _id 冲突 → Err
332        assert!(insert_one(&mut r, &json!({ "_id": "a", "n": 2 })).is_err());
333        assert_eq!(r.len(), 1);
334        // null doc → 空对象(不报错)
335        assert_eq!(insert_one(&mut r, &Value::Null).unwrap(), json!({}));
336        // 非对象 → Err
337        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        // 与既有冲突
344        assert!(insert_many(&mut r, &[json!({ "_id": "a" })], false).is_err());
345        // 批内重复
346        assert!(insert_many(
347            &mut r,
348            &[json!({ "_id": "b" }), json!({ "_id": "b" })],
349            false
350        )
351        .is_err());
352        // 正常插入
353        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        // 按 _id replace(归档幂等)
367        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        // 重复调用结果稳定
379        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        // $set 同值 → 不计入 modifiedCount
402        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        // 整型 + 整型 → 整型
415        update_many(&mut r, &json!({ "_id": 1 }), &json!({ "$inc": { "i": 3 } })).unwrap();
416        assert_eq!(r[0]["i"], json!(5));
417        // 缺失字段 → 取增量值(保留类型)
418        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        // 非数值目标 → Err
426        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        // 命中 → 返回更新后文档(默认 after)
440        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        // returnDocument=before → 返回更新前
449        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        // 无命中且非 upsert → null
459        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        // upsert:条件等值字段为底 + $set + $setOnInsert
469        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        // 命中 0 → 0
491        assert_eq!(
492            delete_many(&mut r, &json!({ "s": "nope" })).unwrap(),
493            json!({ "deletedCount": 0 })
494        );
495    }
496}