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

rust_store_core/local/
eval.rs
