rust_store_core/local/
eval.rs

1//! 本地磁盘数据源 —— 命令分发（core 规划产出的 Mongo 命令 JSON → 求值）。
2//!
3//! 语义基准 = **MongoDB 驱动语义**：本层与 `mongo` 路径吃**同一份**命令、必须同结果。
4//! 纯逻辑：无 IO / 无时钟 / 无随机 ——「整目录集合快照」由宿主提供并回收（文件读写只在宿主）。
5//!
6//! 返回契约（执行文档 §4.1）：
7//! `{ "result": <驱动等价返回值>, "changed": ["<集合名>", …], "collections": {…} }`
8//!   - 读命令（`find` / `aggregate` / `countDocuments` / `findOne`）→ `changed` 恒 `[]`，
9//!     `collections` 原样回传；
10//!   - 写命令 → `collections` 为变更后快照、`changed` 为被改集合名。
11//!
12//! 单集合文档数超护栏 [`MAX_LOCAL_COLLECTION_DOCS`] → `Err`（禁静默截断）。
13
14use serde_json::{json, Map, Value};
15
16use crate::local::value::{apply_projection, MAX_LOCAL_COLLECTION_DOCS};
17use crate::local::{filter, pipeline, update};
18
19/// 本地磁盘数据源单命令求值（纯逻辑，无 IO / 无时钟 / 无随机）。
20///
21/// `collections`：`{ "<物理集合名>": [文档, …] }` —— 宿主提供的**整目录快照**；
22///                命令 `$lookup.from` 指向的集合必须也在其中（缺集合按空集合处理）。
23/// `command`：core 规划产出的 Mongo 命令 JSON（与 mongo 路径**同一份**）。
24pub fn eval_command(collections: &Value, command: &Value) -> Result<Value, String> {
25    let cmd = command
26        .as_object()
27        .ok_or_else(|| format!("command 必须是对象，收到 {}", type_name(command)))?;
28    let kind = cmd
29        .get("kind")
30        .and_then(Value::as_str)
31        .ok_or_else(|| "command 缺少 kind".to_string())?;
32    let name = cmd
33        .get("collection")
34        .and_then(Value::as_str)
35        .ok_or_else(|| "command 缺少 collection".to_string())?;
36    let mut map = collections
37        .as_object()
38        .cloned()
39        .ok_or_else(|| format!("collections 必须是对象 {{集合名: [文档…]}}，收到 {}", type_name(collections)))?;
40
41    // ── 读命令：changed 恒空、collections 原样回传 ──────────────────────────
42    match kind {
43        "find" => {
44            let rows = rows_of(&map, name)?;
45            let condition = cmd.get("filter").cloned().unwrap_or(Value::Null);
46            let projection = cmd.get("projection").cloned().unwrap_or(Value::Null);
47            let mut out = Vec::new();
48            for row in &rows {
49                if filter::matches(row, &condition)? {
50                    out.push(project(row, &projection)?);
51                }
52            }
53            return Ok(envelope(Value::Array(out), Vec::new(), collections));
54        }
55        "aggregate" => {
56            let stages = cmd
57                .get("pipeline")
58                .and_then(Value::as_array)
59                .ok_or_else(|| "aggregate 需要 pipeline 数组".to_string())?;
60            let out = pipeline::run(collections, name, stages)?;
61            return Ok(envelope(Value::Array(out), Vec::new(), collections));
62        }
63        "countDocuments" => {
64            let rows = rows_of(&map, name)?;
65            let condition = cmd.get("filter").cloned().unwrap_or(Value::Null);
66            let mut n = 0_usize;
67            for row in &rows {
68                if filter::matches(row, &condition)? {
69                    n += 1;
70                }
71            }
72            return Ok(envelope(json!(n), Vec::new(), collections));
73        }
74        "findOne" => {
75            let rows = rows_of(&map, name)?;
76            let condition = cmd.get("filter").cloned().unwrap_or(Value::Null);
77            let projection = cmd.get("projection").cloned().unwrap_or(Value::Null);
78            let mut found = Value::Null;
79            for row in &rows {
80                if filter::matches(row, &condition)? {
81                    found = project(row, &projection)?;
82                    break;
83                }
84            }
85            return Ok(envelope(found, Vec::new(), collections));
86        }
87        _ => {}
88    }
89
90    // ── 写命令：变更后快照 + 非空 changed ──────────────────────────────────
91    let mut rows = rows_of(&map, name)?;
92    let result = match kind {
93        "insertOne" => {
94            let doc = cmd
95                .get("doc")
96                .ok_or_else(|| "insertOne 缺少 doc".to_string())?;
97            update::insert_one(&mut rows, doc)?
98        }
99        "insertMany" => {
100            let docs = cmd
101                .get("docs")
102                .and_then(Value::as_array)
103                .ok_or_else(|| "insertMany 需要 docs 数组".to_string())?;
104            let upsert_by_id = cmd
105                .get("upsertById")
106                .and_then(Value::as_bool)
107                .unwrap_or(false);
108            update::insert_many(&mut rows, docs, upsert_by_id)?
109        }
110        "updateMany" => {
111            let condition = cmd.get("filter").cloned().unwrap_or(Value::Null);
112            let spec = cmd
113                .get("update")
114                .ok_or_else(|| "updateMany 缺少 update".to_string())?;
115            update::update_many(&mut rows, &condition, spec)?
116        }
117        "findOneAndUpdate" => {
118            let condition = cmd.get("filter").cloned().unwrap_or(Value::Null);
119            let spec = cmd
120                .get("update")
121                .ok_or_else(|| "findOneAndUpdate 缺少 update".to_string())?;
122            let options = cmd.get("options").cloned().unwrap_or(Value::Null);
123            update::find_one_and_update(&mut rows, &condition, spec, &options)?
124        }
125        "deleteMany" => {
126            let condition = cmd.get("filter").cloned().unwrap_or(Value::Null);
127            update::delete_many(&mut rows, &condition)?
128        }
129        other => {
130            return Err(format!(
131                "未支持的命令: {other}（本地求值器拒绝静默；支持 find/aggregate/countDocuments/findOne/insertOne/insertMany/updateMany/findOneAndUpdate/deleteMany）"
132            ));
133        }
134    };
135
136    guard_docs(name, &rows)?;
137    map.insert(name.to_string(), Value::Array(rows));
138    Ok(envelope(result, vec![name.to_string()], &Value::Object(map)))
139}
140
141/// 返回包络：`{ result, changed, collections }`。
142fn envelope(result: Value, changed: Vec<String>, collections: &Value) -> Value {
143    json!({
144        "result": result,
145        "changed": changed,
146        "collections": collections,
147    })
148}
149
150/// 取集合文档数组；缺失 → 空数组；非数组 → `Err`（禁静默当空集合）。
151fn rows_of(map: &Map<String, Value>, name: &str) -> Result<Vec<Value>, String> {
152    match map.get(name) {
153        None => Ok(Vec::new()),
154        Some(Value::Array(a)) => Ok(a.clone()),
155        Some(other) => Err(format!(
156            "集合 \"{name}\" 必须是文档数组，收到 {}",
157            type_name(other)
158        )),
159    }
160}
161
162/// 投影：`null`（未声明）→ 原样；否则按 [`apply_projection`]。
163fn project(doc: &Value, projection: &Value) -> Result<Value, String> {
164    if projection.is_null() {
165        return Ok(doc.clone());
166    }
167    apply_projection(doc, projection)
168}
169
170/// 文档数护栏（禁静默截断）。
171fn guard_docs(name: &str, rows: &[Value]) -> Result<(), String> {
172    if rows.len() > MAX_LOCAL_COLLECTION_DOCS {
173        return Err(format!(
174            "集合 \"{name}\" 文档数 {} 超过护栏 MAX_LOCAL_COLLECTION_DOCS={}（拒绝静默截断）",
175            rows.len(),
176            MAX_LOCAL_COLLECTION_DOCS
177        ));
178    }
179    Ok(())
180}
181
182fn type_name(v: &Value) -> &'static str {
183    match v {
184        Value::Null => "null",
185        Value::Bool(_) => "bool",
186        Value::Number(_) => "number",
187        Value::String(_) => "string",
188        Value::Array(_) => "array",
189        Value::Object(_) => "object",
190    }
191}
192
193#[cfg(test)]
194mod tests {
195    use super::*;
196    use serde_json::json;
197
198    #[test]
199    fn read_commands_have_empty_changed_and_passthrough() {
200        let collections = json!({ "users": [{"_id": "u1", "name": "Ada", "age": 36}] });
201        // find
202        let out = eval_command(
203            &collections,
204            &json!({ "kind": "find", "collection": "users", "filter": { "age": { "$gt": 30 } } }),
205        )
206        .unwrap();
207        assert_eq!(out["result"], json!([{"_id": "u1", "name": "Ada", "age": 36}]));
208        assert_eq!(out["changed"], json!([]));
209        assert_eq!(out["collections"], collections);
210        // countDocuments
211        let c = eval_command(
212            &collections,
213            &json!({ "kind": "countDocuments", "collection": "users", "filter": null }),
214        )
215        .unwrap();
216        assert_eq!(c["result"], json!(1));
217        assert_eq!(c["changed"], json!([]));
218        // findOne 未命中 → null
219        let one = eval_command(
220            &collections,
221            &json!({ "kind": "findOne", "collection": "users", "filter": { "_id": "nope" } }),
222        )
223        .unwrap();
224        assert_eq!(one["result"], Value::Null);
225    }
226
227    #[test]
228    fn find_projection_applied() {
229        let collections = json!({ "users": [{"_id": "u1", "name": "Ada", "age": 36}] });
230        let out = eval_command(
231            &collections,
232            &json!({
233                "kind": "find",
234                "collection": "users",
235                "filter": null,
236                "projection": { "name": 1, "_id": 0 }
237            }),
238        )
239        .unwrap();
240        assert_eq!(out["result"], json!([{"name": "Ada"}]));
241    }
242
243    #[test]
244    fn insert_then_query_roundtrip() {
245        let empty = json!({});
246        let ins = eval_command(
247            &empty,
248            &json!({ "kind": "insertOne", "collection": "users", "doc": { "_id": "u1", "name": "Ada" } }),
249        )
250        .unwrap();
251        assert_eq!(ins["result"], json!({ "_id": "u1", "name": "Ada" }));
252        assert_eq!(ins["changed"], json!(["users"]));
253        let snapshot = ins["collections"].clone();
254        // 用变更后快照续查
255        let q = eval_command(
256            &snapshot,
257            &json!({ "kind": "find", "collection": "users", "filter": { "_id": "u1" } }),
258        )
259        .unwrap();
260        assert_eq!(q["result"], json!([{ "_id": "u1", "name": "Ada" }]));
261    }
262
263    /// 总纲 E4 的关系查询原样命令（aggregate：`$match + $lookup + $unwind + $project`）。
264    #[test]
265    fn aggregate_relation_golden() {
266        let collections = json!({
267            "probeUsers": [
268                {"_id": "pu1", "name": "Ada"},
269                {"_id": "pu2", "name": "Bob"}
270            ],
271            "probePosts": [
272                {"_id": "pp1", "title": "P1", "userId": "pu1"},
273                {"_id": "pp2", "title": "P2", "userId": "pu2"}
274            ]
275        });
276        let command = json!({
277            "kind": "aggregate",
278            "collection": "probePosts",
279            "pipeline": [
280                {"$match": {}},
281                {"$lookup": {
282                    "as": "author",
283                    "from": "probeUsers",
284                    "let": {"rel_userId": {"$ifNull": ["$userId", null]}},
285                    "pipeline": [
286                        {"$match": {"$expr": {"$eq": ["$_id", "$$rel_userId"]}}},
287                        {"$project": {"_id": 1, "name": 1}}
288                    ]
289                }},
290                {"$unwind": {"path": "$author", "preserveNullAndEmptyArrays": true}},
291                {"$project": {"_id": 1, "author": 1, "title": 1}}
292            ]
293        });
294        let out = eval_command(&collections, &command).unwrap();
295        assert_eq!(
296            out["result"],
297            json!([
298                {"_id": "pp1", "author": {"_id": "pu1", "name": "Ada"}, "title": "P1"},
299                {"_id": "pp2", "author": {"_id": "pu2", "name": "Bob"}, "title": "P2"}
300            ])
301        );
302        assert_eq!(out["changed"], json!([]));
303        assert_eq!(out["collections"], collections);
304    }
305
306    #[test]
307    fn write_commands_envelope_and_counts() {
308        let mut snapshot = json!({ "posts": [
309            {"_id": "p1", "title": "P1", "userId": "u1"},
310            {"_id": "p2", "title": "P2", "userId": "u2"}
311        ]});
312        // insertMany
313        let ins = eval_command(
314            &snapshot,
315            &json!({ "kind": "insertMany", "collection": "posts", "docs": [{ "_id": "p3" }] }),
316        )
317        .unwrap();
318        assert_eq!(ins["result"], json!({ "insertedCount": 1 }));
319        assert_eq!(ins["changed"], json!(["posts"]));
320        snapshot = ins["collections"].clone();
321        // updateMany
322        let upd = eval_command(
323            &snapshot,
324            &json!({
325                "kind": "updateMany",
326                "collection": "posts",
327                "filter": { "userId": "u1" },
328                "update": { "$set": { "seen": true } }
329            }),
330        )
331        .unwrap();
332        assert_eq!(upd["result"], json!({ "modifiedCount": 1 }));
333        snapshot = upd["collections"].clone();
334        // deleteMany
335        let del = eval_command(
336            &snapshot,
337            &json!({ "kind": "deleteMany", "collection": "posts", "filter": { "_id": "p3" } }),
338        )
339        .unwrap();
340        assert_eq!(del["result"], json!({ "deletedCount": 1 }));
341    }
342
343    #[test]
344    fn archive_upsert_by_id_through_eval() {
345        let snapshot = json!({ "usersDeleted": [] });
346        let out = eval_command(
347            &snapshot,
348            &json!({
349                "kind": "insertMany",
350                "collection": "usersDeleted",
351                "upsertById": true,
352                "docs": [{ "_id": "u1", "deletedAt": 100 }]
353            }),
354        )
355        .unwrap();
356        assert_eq!(out["result"], json!({ "insertedCount": 1 }));
357        assert_eq!(
358            out["collections"],
359            json!({ "usersDeleted": [{"_id": "u1", "deletedAt": 100}] })
360        );
361    }
362
363    #[test]
364    fn unknown_command_and_collection_type_error() {
365        // 未知命令 → Err
366        assert!(eval_command(&json!({}), &json!({ "kind": "drop", "collection": "t" })).is_err());
367        // 缺少 kind / collection → Err
368        assert!(eval_command(&json!({}), &json!({ "collection": "t" })).is_err());
369        assert!(eval_command(&json!({}), &json!({ "kind": "find" })).is_err());
370        // 集合非数组 → Err（禁静默当空集合）
371        assert!(eval_command(&json!({ "t": 1 }), &json!({ "kind": "find", "collection": "t" })).is_err());
372        // collections 非对象 → Err
373        assert!(eval_command(&json!(5), &json!({ "kind": "find", "collection": "t" })).is_err());
374    }
375
376    #[test]
377    fn guard_docs_over_limit_errors() {
378        // 预置已达护栏的集合，再插一条 → 超限 Err（不构造超大批次，避免冲突检查放大）
379        let docs: Vec<Value> = (0..MAX_LOCAL_COLLECTION_DOCS)
380            .map(|i| json!({ "_id": i }))
381            .collect();
382        let snapshot = json!({ "big": docs });
383        let command = json!({ "kind": "insertOne", "collection": "big", "doc": { "_id": "x" } });
384        let err = eval_command(&snapshot, &command).unwrap_err();
385        assert!(err.contains("MAX_LOCAL_COLLECTION_DOCS"), "err = {err}");
386    }
387}
rust_store_core/local/eval.rs

rust_store_core/local/
eval.rs
