Skip to main content

rust_store_core/schema/
registry.rs

1//! Schema 注册表:注册/定位/配置(数据结构与规范化见 [`super::definition`])。
2
3use std::collections::HashMap;
4
5use serde_json::{json, Map, Value};
6
7use crate::types::{is_truthy, str_list, AGG_OPS};
8
9use super::definition::{normalize_fields, ComputeDef, FieldDef, Location, RelationDef, Schema};
10
11use crate::command::WriteLinkPolicy;
12
13/// 查询档位:判决唯一在 core(照 [`Registry::require_context`] 既有范式)。
14///
15/// - [`Profile::Standard`](默认):标准调用 —— 跨方言对齐的公共能力集;
16///   DB 独有能力可用但须代码注释标注「不建议业务查询」(见执行文档 §4.5)。
17/// - [`Profile::Text2Query`]:AI 问数 —— 功能收缩 + 硬性限制(行数/深度/准入全收紧)。
18#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
19pub enum Profile {
20    /// 标准调用:跨方言对齐的公共能力集;DB 独有能力可用但须注释标注(不建议业务查询)
21    #[default]
22    Standard,
23    /// AI 问数:功能收缩 + 硬性限制
24    Text2Query,
25}
26
27impl Profile {
28    /// 档位字符串(Host / 绑定层单点取用)
29    pub fn as_str(&self) -> &'static str {
30        match self {
31            Profile::Standard => "standard",
32            Profile::Text2Query => "text2query",
33        }
34    }
35
36    /// 字符串 → 档位;未知值 **Err**(禁静默回落 `Standard`,见执行文档 §7)
37    pub fn from_str_or_err(s: &str) -> Result<Profile, String> {
38        match s {
39            "standard" => Ok(Profile::Standard),
40            "text2query" => Ok(Profile::Text2Query),
41            other => Err(format!("未知 profile: {other}(仅 standard / text2query)")),
42        }
43    }
44}
45
46/// 一条注册项:一份主结构 + 主/从链路 + 版本(D4/D5/D13)。
47#[derive(Debug, Clone)]
48struct Entry {
49    schema: Schema,
50    /// 主 + 从链路;`links[0]` 恒为主落点(= `schema` 的定位投影)
51    links: Vec<Location>,
52    /// 版本:首次注册 = 1;跨批次 reload 同名 ⇒ +1
53    version: u64,
54}
55
56/// 批次分组结果(判据单点输出;`register_batch` 与 `schema::load::plan_load` 共用)。
57pub(super) struct BatchClass {
58    /// 组名(按首次出现顺序)
59    pub order: Vec<String>,
60    /// name → (主定义, 落点, collection);每组至多一份主
61    pub primaries: HashMap<String, (Value, Location, String)>,
62    /// name → 从项(原序)
63    pub replicas: HashMap<String, Vec<(Value, Location)>>,
64}
65
66/// 批次判据错误(调用方各自格式化文案:注册期与装载期文案口径不同)。
67pub(super) enum BatchError {
68    /// 解析错误(非对象 / 缺 name)
69    Parse(String),
70    /// 同名主 ≥2(A1):组名 + 主项在输入中的下标
71    PrimaryDuplicate { name: String, indices: Vec<usize> },
72    /// 批内四元组冲突
73    LocationConflict(Box<LocationConflict>),
74}
75
76/// 批次判据单点:**分组 + 主唯一(≥2 ⇒ `Err`)+ 批内四元组冲突(⇒ `Err`)**。
77///
78/// **0 主不在此报错**(交由调用方决定:`register_batch` 走既有 entry 追加从链路,
79/// `plan_load` 直接拒——「主定义缺失」)。`register_batch` 与 `plan_load` 共用本函数,
80/// 禁各写一份(判据单点,防 A1/A3 漂移)。
81pub(super) fn classify_batch(items: &[(Value, Location)]) -> Result<BatchClass, BatchError> {
82    let mut order: Vec<String> = Vec::new();
83    let mut primaries: HashMap<String, (Value, Location, String)> = HashMap::new();
84    let mut primary_idx: HashMap<String, usize> = HashMap::new();
85    let mut replicas: HashMap<String, Vec<(Value, Location)>> = HashMap::new();
86
87    for (idx, (defn, loc)) in items.iter().enumerate() {
88        let obj = defn
89            .as_object()
90            .ok_or_else(|| BatchError::Parse("schema 定义必须是对象".to_string()))?;
91        let name = obj
92            .get("name")
93            .and_then(|v| v.as_str())
94            .ok_or_else(|| BatchError::Parse("schema 缺少 name".to_string()))?
95            .to_string();
96        if !order.iter().any(|n| n == &name) {
97            order.push(name.clone());
98        }
99
100        if obj.get("replica").map(is_truthy).unwrap_or(false) {
101            replicas
102                .entry(name)
103                .or_default()
104                .push((defn.clone(), loc.clone()));
105            continue;
106        }
107
108        let collection = obj
109            .get("collection")
110            .and_then(|v| v.as_str())
111            .filter(|s| !s.is_empty())
112            .map(String::from)
113            .unwrap_or_else(|| name.clone());
114        if let Some(first) = primary_idx.get(&name) {
115            // 同名主 ≥2 ⇒ Err(A1);携带下标供调用方定位(rel 路径 / 提示)
116            return Err(BatchError::PrimaryDuplicate {
117                name,
118                indices: vec![*first, idx],
119            });
120        }
121        primary_idx.insert(name.clone(), idx);
122        primaries.insert(name, (defn.clone(), loc.clone(), collection));
123    }
124
125    // 批内四元组冲突:主 + 从全部链路(判据函数 `detect_location_conflict`)
126    let mut quads: Vec<(String, String, Location)> = Vec::new();
127    for name in &order {
128        if let Some((_, loc, coll)) = primaries.get(name) {
129            quads.push((name.clone(), coll.clone(), loc.clone()));
130        }
131        if let Some(rs) = replicas.get(name) {
132            for (defn, loc) in rs {
133                let coll = defn
134                    .get("name")
135                    .and_then(|v| v.as_str())
136                    .unwrap_or(name)
137                    .to_string();
138                quads.push((name.clone(), coll, loc.clone()));
139            }
140        }
141    }
142    if let Some(c) = detect_location_conflict(&quads) {
143        return Err(BatchError::LocationConflict(Box::new(c)));
144    }
145
146    Ok(BatchClass {
147        order,
148        primaries,
149        replicas,
150    })
151}
152
153/// 四元组冲突信息(供调用方格式化文案)
154pub(super) struct LocationConflict {
155    pub collection: String,
156    pub prev: String,
157    pub name: String,
158    pub source: String,
159    pub database: Option<String>,
160    pub schema: Option<String>,
161}
162
163/// 四元组 `(source, database, schema, collection)` 冲突检测(**判据单点**):
164/// 同一四元组被两个不同 schema 名占用 ⇒ `Some(冲突)`。`register_batch`(含既有 entry)
165/// 与 `plan_load`(批内)共用。
166pub(super) fn detect_location_conflict(
167    entries: &[(String, String, Location)],
168) -> Option<LocationConflict> {
169    let mut seen: HashMap<(String, Option<String>, Option<String>, String), String> =
170        HashMap::new();
171    for (name, collection, loc) in entries {
172        let key = (
173            loc.source.clone(),
174            loc.database.clone(),
175            loc.schema.clone(),
176            collection.clone(),
177        );
178        if let Some(prev) = seen.get(&key) {
179            if prev != name {
180                return Some(LocationConflict {
181                    collection: collection.clone(),
182                    prev: prev.clone(),
183                    name: name.clone(),
184                    source: loc.source.clone(),
185                    database: loc.database.clone(),
186                    schema: loc.schema.clone(),
187                });
188            }
189        } else {
190            seen.insert(key, name.clone());
191        }
192    }
193    None
194}
195
196/// `BatchError` → 注册期文案(保留既有子串:`同名主定义重复` / `定位冲突`)。
197fn batch_err_to_msg(e: BatchError) -> String {
198    match e {
199        BatchError::Parse(m) => m,
200        BatchError::PrimaryDuplicate { name, .. } => format!(
201            "同名主定义重复: {}(同一装载批次内 name 必须唯一,从定义请用 replica: true)",
202            name
203        ),
204        BatchError::LocationConflict(c) => format!(
205            "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
206            c.source, c.database, c.schema, c.collection, c.prev
207        ),
208    }
209}
210
211#[derive(Debug, Clone, Default)]
212pub struct Registry {
213    entries: HashMap<String, Entry>,
214    order: Vec<String>,
215    /// 上下文强制开关(默认关闭 = fail-open,与 JS 原版 parity);
216    /// 开启后所有 plan 入口对 `ctx: None` 显式报错(fail-secure,见 `permission` 模块文档)。
217    /// 内部调用请传显式系统上下文(JSON `{"internal": true}` / `Context::system()`)。
218    require_context: bool,
219    /// 查询档位(默认 [`Profile::Standard`];text2query 由 AI 问数链路显式进入)。
220    profile: Profile,
221    /// RBAC 动态策略(`None` = 未启用,判决原语直通、行为与现状一致)。见 [`crate::rbac`]
222    rbac: Option<crate::rbac::RbacPolicy>,
223    /// 角色清单与未配置姿态(豁免 / 拒写 / Open|Closed,默认 []/[]/Open)。见 [`crate::permission::RoleRules`]
224    role_rules: crate::permission::RoleRules,
225    /// 定义层门禁策略(默认 Open —— 全放行,保持既有 parity)。见 [`crate::permission::MetaPolicy`]
226    meta_policy: crate::permission::MetaPolicy,
227    /// 跨连接写策略(默认 [`WriteLinkPolicy::Reject`])。见 [`crate::command::write_links`]
228    write_link_policy: WriteLinkPolicy,
229}
230
231impl Registry {
232    pub fn new() -> Self {
233        Self::default()
234    }
235
236    /// 注册一个 schema(含自动注册 `<Name>Deleted` 归档表),对应 JS `register`。
237    ///
238    /// 兼容入口:单条 + 缺省落点(`source="default"`、`database/schema=None`)。
239    /// 定义层门禁按当前 [`MetaPolicy`](crate::permission::MetaPolicy) 判决
240    /// (默认 Open → 全放行)。需带 ctx 显式过门禁请用 [`Self::register_with_ctx`]。
241    pub fn register(&mut self, defn: &Value) -> Result<(), String> {
242        self.register_with_ctx(defn, None)
243    }
244
245    /// 带 ctx 的注册(单条 + 缺省落点):内部转 [`Self::register_batch`]。
246    pub fn register_with_ctx(
247        &mut self,
248        defn: &Value,
249        ctx: Option<&crate::permission::Context>,
250    ) -> Result<(), String> {
251        self.register_batch(&[(defn.clone(), Location::default())], ctx)
252    }
253
254    /// 带定位的批量注册(D13:唯一性校验落「装载批次集合」)。
255    ///
256    /// 算法(**先构造后落库,任一步 Err ⇒ 零副作用**):
257    /// 1. 判据单点 [`classify_batch`](分组 / 主唯一 ≥2 ⇒ `Err` / 批内四元组冲突),与
258    ///    `schema::load::plan_load` 同源。
259    /// 2. `can_register` 门禁(判决先于 `build_schema`)。
260    /// 3. 逐组:有主 ⇒ `build_schema` 为 **主**;0 主 ⇒ 追加既有同名 entry 的从链路,无既有 ⇒ `Err`。
261    /// 4. 全批 + 既有 `(source, database, schema, collection)` 跨名冲突 ⇒ `Err`
262    ///    (判据函数 [`detect_location_conflict`],替代旧 `check_location_unique`)。
263    /// 5. 提交:`links = [primary.loc] ++ replicas`;命中既有 entry ⇒ `version + 1`
264    ///    (`order` 位置不变)、替换结构与链路;新名 ⇒ `version = 1`、`order.push`。
265    ///    `Schema.source/database/schema` 由 `primary.loc` 注入(`source == "default"` ⇒ `None`)。
266    /// 6. 归档派生:主 `name` 不以 `"Deleted"` 结尾且非 `_isArchive` ⇒ 派生 `<Name>Deleted`
267    ///    (`links` 继承主链路,定位同主)。
268    pub fn register_batch(
269        &mut self,
270        items: &[(Value, Location)],
271        ctx: Option<&crate::permission::Context>,
272    ) -> Result<(), String> {
273        // ── 1:判据单点(分组 / 主唯一 / 批内四元组冲突),与 plan_load 同源 ──
274        let class = classify_batch(items).map_err(batch_err_to_msg)?;
275
276        // 定义层门禁:判决先于 build_schema —— 拒绝即返回,绝不部分写入
277        for (defn, _loc) in items {
278            let name = defn.get("name").and_then(|v| v.as_str()).unwrap_or("");
279            if !crate::permission::can_register(&self.meta_policy, ctx) {
280                return Err(format!("ERR_PERMISSION: 无权注册或覆盖定义 {name}"));
281            }
282        }
283
284        // ── 2:逐组拟定 entry 序列(主 + 归档附表,保序) ──
285        struct Planned {
286            name: String,
287            schema: Schema,
288            links: Vec<Location>,
289        }
290        let mut planned: Vec<Planned> = Vec::new();
291        for name in &class.order {
292            let rs: Vec<Location> = class
293                .replicas
294                .get(name)
295                .map(|v| v.iter().map(|(_, l)| l.clone()).collect())
296                .unwrap_or_default();
297            if let Some((raw, loc, collection)) = class.primaries.get(name) {
298                let obj = raw.as_object().unwrap();
299                let schema = build_schema(obj, name.clone(), collection.clone(), loc)?;
300                let is_archive = obj.get("_isArchive").map(is_truthy).unwrap_or(false);
301                let mut links = vec![loc.clone()];
302                links.extend(rs);
303                planned.push(Planned {
304                    name: name.clone(),
305                    schema: schema.clone(),
306                    links: links.clone(),
307                });
308                // 归档附表紧随主定义派生(`links` 继承主链路,定位同主)
309                if !is_archive && !name.ends_with("Deleted") {
310                    let aobj = archive_defn(obj, name, &schema.collection)
311                        .as_object()
312                        .cloned()
313                        .unwrap();
314                    let aname = format!("{}Deleted", name);
315                    let acoll = format!("{}_deleted", schema.collection);
316                    let aschema = build_schema(&aobj, aname.clone(), acoll, &links[0])?;
317                    planned.push(Planned {
318                        name: aname,
319                        schema: aschema,
320                        links,
321                    });
322                }
323            } else {
324                // 0 主:只能是对既有 entry 追加从链路(跨批演进)
325                let Some(existing) = self.entries.get(name) else {
326                    return Err(format!(
327                        "未找到主 schema: {}(从定义 `replica` 必须伴随同批主定义,或该 name 已注册)",
328                        name
329                    ));
330                };
331                let mut links = existing.links.clone();
332                links.extend(rs);
333                planned.push(Planned {
334                    name: name.clone(),
335                    schema: existing.schema.clone(),
336                    links,
337                });
338            }
339        }
340
341        // ── 3:四元组唯一性(本批 + 既有,跨名冲突即 Err;判据函数单点) ──
342        let batch_names: std::collections::HashSet<&str> =
343            planned.iter().map(|p| p.name.as_str()).collect();
344        let mut quads: Vec<(String, String, Location)> = Vec::new();
345        for (n, e) in &self.entries {
346            if batch_names.contains(n.as_str()) {
347                continue; // 本批将整体替换该名,旧链路不参与冲突判定
348            }
349            for l in &e.links {
350                quads.push((n.clone(), e.schema.collection.clone(), l.clone()));
351            }
352        }
353        for p in &planned {
354            for l in &p.links {
355                quads.push((p.name.clone(), p.schema.collection.clone(), l.clone()));
356            }
357        }
358        if let Some(c) = detect_location_conflict(&quads) {
359            return Err(format!(
360                "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
361                c.source, c.database, c.schema, c.collection, c.prev
362            ));
363        }
364
365        // ── 4:提交(命中既有 ⇒ version + 1 且 order 位置不变;新名 ⇒ version = 1 + push) ──
366        for p in planned {
367            if let Some(e) = self.entries.get_mut(&p.name) {
368                e.schema = p.schema;
369                e.links = p.links;
370                e.version += 1;
371            } else {
372                self.entries.insert(
373                    p.name.clone(),
374                    Entry {
375                        schema: p.schema,
376                        links: p.links,
377                        version: 1,
378                    },
379                );
380                self.order.push(p.name);
381            }
382        }
383
384        Ok(())
385    }
386
387    /// 主链路 + 从链路(只读)
388    pub fn links_of(&self, name: &str) -> Result<&[Location], String> {
389        Ok(&self.get_entry(name)?.links)
390    }
391
392    /// 主落点(= `links[0]`)
393    pub fn primary_location(&self, name: &str) -> Result<&Location, String> {
394        self.get_entry(name)?
395            .links
396            .first()
397            .ok_or_else(|| format!("Schema 无落点链路: {}", name))
398    }
399
400    /// 版本(首次 = 1)
401    pub fn version_of(&self, name: &str) -> Result<u64, String> {
402        Ok(self.get_entry(name)?.version)
403    }
404
405    fn get_entry(&self, name: &str) -> Result<&Entry, String> {
406        self.entries
407            .get(name)
408            .ok_or_else(|| format!("Schema 未注册: {}", name))
409    }
410
411    /// 按名称获取 schema
412    pub fn get(&self, name: &str) -> Result<&Schema, String> {
413        self.get_entry(name).map(|e| &e.schema)
414    }
415
416    pub fn has(&self, name: &str) -> bool {
417        self.entries.contains_key(name)
418    }
419
420    /// 清空 schema 注册表(`schemas` + `order`),对应绑定层的测试隔离 / 动态重建场景。
421    ///
422    /// 只清 schema,**不动**配置开关(`require_context` / `profile` / `rbac` /
423    /// `role_rules`)与回调表(`clear_fns` 对称:各清各的)——开关生命周期属
424    /// Registry 配置面,不随 schema 集合重建而丢。
425    pub fn clear(&mut self) {
426        self.entries.clear();
427        self.order.clear();
428    }
429
430    /// 开关「上下文强制」(默认关闭 = fail-open,保持 JS parity)。
431    /// 开启后:plan 入口遇 `ctx: None` 报 `ERR_NO_CONTEXT`(fail-secure)。
432    /// 内部调用须显式传系统上下文(`{"internal": true}`)。
433    pub fn set_require_context(&mut self, require: bool) {
434        self.require_context = require;
435    }
436
437    /// 「上下文强制」开关当前值
438    pub fn require_context(&self) -> bool {
439        self.require_context
440    }
441
442    /// 设置查询档位(`standard` / `text2query`)。
443    ///
444    /// 档位为**单值状态**(非栈):由 Host 的 `text2query()` 上下文管理器负责
445    /// 进入时设档、退出时恢复(见 `py-store` / `nodejs-store` 门面)。判决一律在 core。
446    pub fn set_profile(&mut self, profile: Profile) {
447        self.profile = profile;
448    }
449
450    /// 当前查询档位
451    pub fn profile(&self) -> Profile {
452        self.profile
453    }
454
455    /// 豁免角色清单:命中者在一切判决环节(静态 + RBAC)直接放行。默认空——无豁免。
456    pub fn set_exempt_roles(&mut self, roles: Vec<String>) {
457        self.role_rules.exempt_roles = roles;
458    }
459
460    /// 拒写角色清单:命中者一切写路径拒绝(读不受影响)。默认空——无拒写。
461    pub fn set_deny_write_roles(&mut self, roles: Vec<String>) {
462        self.role_rules.deny_write_roles = roles;
463    }
464
465    /// schema 白名单缺失/为空时的默认姿态。默认 Open(保持现状语义)。
466    pub fn set_unconfigured_policy(&mut self, policy: crate::permission::UnconfiguredPolicy) {
467        self.role_rules.unconfigured = policy;
468    }
469
470    /// 当前角色规则(静态判决函数与 RBAC decide 的共用取参入口)
471    pub fn role_rules(&self) -> &crate::permission::RoleRules {
472        &self.role_rules
473    }
474
475    /// 定义层门禁策略(`closed=true` 时仅 internal 或 `roles` 白名单可注册/覆盖)。
476    /// 默认 Open —— 全放行(保持既有 parity;须在首次业务注册前调用方生效于该次注册)。
477    pub fn set_meta_policy(&mut self, closed: bool, roles: Vec<String>) {
478        self.meta_policy = crate::permission::MetaPolicy { closed, roles };
479    }
480
481    /// 当前定义层门禁策略
482    pub fn meta_policy(&self) -> &crate::permission::MetaPolicy {
483        &self.meta_policy
484    }
485
486    /// 设置跨连接写策略(默认 [`WriteLinkPolicy::Reject`])。
487    ///
488    /// 判决唯一在 core([`crate::command::write_links::resolve_write_links`]);
489    /// 宿主仅需在初始化期把它透传到 registry。
490    pub fn set_write_link_policy(&mut self, policy: WriteLinkPolicy) {
491        self.write_link_policy = policy;
492    }
493
494    /// 当前跨连接写策略
495    pub fn write_link_policy(&self) -> WriteLinkPolicy {
496        self.write_link_policy
497    }
498
499    /// 只读定义层判决(`workflow` 等宿主侧定义面复用同一门禁;判决唯一在 core)。
500    ///
501    /// 不改变任何注册状态:Open 全放行;Closed 仅 `internal` 或 `roles` 白名单角色。
502    /// 语义与 [`Self::register_with_ctx`] 的门禁判据**完全一致**(同一
503    /// [`can_register`](crate::permission::can_register)),仅作只读暴露。
504    pub fn can_register(&self, ctx: Option<&crate::permission::Context>) -> bool {
505        crate::permission::can_register(&self.meta_policy, ctx)
506    }
507
508    /// 注入/清除 RBAC 动态策略;`None` = 关闭(判决原语直通)。
509    /// 解析失败 Err(fail-fast,禁静默吞配置错误),成功后判决链路即刻生效。
510    pub fn set_rbac(&mut self, policy: Option<&Value>) -> Result<(), String> {
511        match policy {
512            None => {
513                self.rbac = None;
514                Ok(())
515            }
516            Some(v) => {
517                let p = crate::rbac::RbacPolicy::from_json(v)?;
518                self.rbac = Some(p);
519                Ok(())
520            }
521        }
522    }
523
524    /// 当前 RBAC 策略(`None` = 未启用)
525    pub fn rbac(&self) -> Option<&crate::rbac::RbacPolicy> {
526        self.rbac.as_ref()
527    }
528
529    /// 按定位四元组精确获取 schema(命令路由的唯一定位入口)
530    ///
531    /// 命中判据:`collection` 相等 **且** 四元组命中该 entry 的**任一链路**。
532    /// **不做任何猜测回落**。
533    pub fn get_by_location(
534        &self,
535        source: &str,
536        database: Option<&str>,
537        schema: Option<&str>,
538        collection: &str,
539    ) -> Result<&Schema, String> {
540        let db = database.filter(|s| !s.is_empty());
541        let sc = schema.filter(|s| !s.is_empty());
542        self.entries
543            .values()
544            .find(|e| {
545                e.schema.collection == collection
546                    && e.links.iter().any(|l| {
547                        l.source == source
548                            && l.database.as_deref().filter(|s| !s.is_empty()) == db
549                            && l.schema.as_deref().filter(|s| !s.is_empty()) == sc
550                    })
551            })
552            .map(|e| &e.schema)
553            .ok_or_else(|| {
554                format!(
555                    "Schema 未定位(source = {}, database = {:?}, schema = {:?}, collection = {})",
556                    source, db, sc, collection
557                )
558            })
559    }
560
561    /// 命令结构定位(方言翻译用):优先四元组精确命中;未命中时(`route_override`
562    /// 改写落点的多租户场景)回落 `(source, collection)` 定位 —— 结构 schema
563    /// 与定位落点正交(见 multi-datasource-routing-plan.md §6:权限/字段校验仍按结构 schema)。
564    ///
565    /// `(source, collection)` 恰一候选 → 命中;多候选时取 `database`/`schema` 均 `null` 的
566    /// 结构声明,无该声明 → 报错(拒绝猜测,铁律 3)。
567    pub fn get_for_command(
568        &self,
569        source: &str,
570        database: Option<&str>,
571        schema: Option<&str>,
572        collection: &str,
573    ) -> Result<&Schema, String> {
574        if let Ok(s) = self.get_by_location(source, database, schema, collection) {
575            return Ok(s);
576        }
577        let candidates: Vec<&Entry> = self
578            .entries
579            .values()
580            .filter(|e| {
581                e.schema.collection == collection && e.links.iter().any(|l| l.source == source)
582            })
583            .collect();
584        match candidates.len() {
585            0 => self.get_by_location(source, database, schema, collection),
586            1 => Ok(&candidates[0].schema),
587            _ => candidates
588                .iter()
589                .find(|e| e.schema.database().is_none() && e.schema.schema().is_none())
590                .map(|e| &e.schema)
591                .ok_or_else(|| {
592                    format!(
593                        "命令定位 ({}, {:?}, {:?}, {}) 存在多个结构 schema(多租户 override 需唯一的 (source, collection) 结构声明)",
594                        source, database, schema, collection
595                    )
596                }),
597        }
598    }
599
600    pub fn list(&self) -> Vec<String> {
601        self.order.clone()
602    }
603
604    /// schema 绑定的数据源名(= 主链路 source;`default` → `None`)
605    pub fn schema_datasource(&self, name: &str) -> Result<Option<String>, String> {
606        Ok(self.get(name)?.source.clone())
607    }
608
609    /// 解析 schema 绑定的数据源(`config` 为 `{ "sources": { name: kind } }`)
610    ///
611    /// 纯逻辑:不持有连接(连接句柄留在 Host)。`null` / 空配置下等价单源 Mongo,
612    /// 保证既有调用方与 parity fixture 零变更。见 [`crate::datasource`]。
613    pub fn resolve_datasource(
614        &self,
615        schema_name: &str,
616        config: &Value,
617    ) -> Result<crate::datasource::DataSource, String> {
618        let declared = self.get(schema_name)?.source.as_deref();
619        let cfg = crate::datasource::DataSourceConfig::from_json(config)?;
620        cfg.resolve(declared)
621    }
622}
623
624/// 由原始定义构建 [`Schema`](字段/关系/计算列/元数据分块解析,保持单一职责)。
625/// 落点(`source`/`database`/`schema`)由 `loc` 注入(定义文件零落点)。
626fn build_schema(
627    obj: &Map<String, Value>,
628    name: String,
629    collection: String,
630    loc: &Location,
631) -> Result<Schema, String> {
632    let timestamps = parse_timestamps(obj)?;
633    let mut fields = normalize_fields(obj.get("fields"))?;
634    if timestamps {
635        add_timestamp_fields(&mut fields);
636    }
637    let relations = parse_relations(obj);
638    // 计算列的 agg 需要校验「引用的关系存在」→ 必须晚于 relations 解析
639    let computes = parse_computes(obj, &relations)?;
640    // 写触发器声明(onFields 校验依赖已归一 fields;非法声明注册期 Err)
641    let triggers = super::triggers::parse_triggers(obj, &name, &fields)?;
642    // 命名冲突检测(A7):fields ∪ relations ∪ computes 归一后撞名 / 撞保留键 ⇒ 报错
643    // (三者同属结果文档命名空间;`_id` 已在 detect_conflicts 内豁免)
644    let mut ids: Vec<&str> = Vec::new();
645    ids.extend(fields.keys().map(String::as_str));
646    ids.extend(relations.keys().map(String::as_str));
647    ids.extend(computes.iter().map(|(k, _)| k.as_str()));
648    crate::naming::detect_conflicts(ids)?;
649    Ok(Schema {
650        name,
651        collection,
652        id_prefix: obj
653            .get("idPrefix")
654            .and_then(|v| v.as_str())
655            .unwrap_or("")
656            .to_string(),
657        timestamps,
658        fields,
659        relations,
660        computes,
661        triggers,
662        read: str_list(obj.get("read")),
663        write: str_list(obj.get("write")),
664        indexes: obj
665            .get("indexes")
666            .and_then(|v| v.as_array())
667            .cloned()
668            .unwrap_or_default(),
669        source: if loc.source == crate::datasource::DEFAULT_SOURCE {
670            None
671        } else {
672            Some(loc.source.clone())
673        },
674        database: loc.database.clone().filter(|s| !s.is_empty()),
675        schema: loc.schema.clone().filter(|s| !s.is_empty()),
676    })
677}
678
679/// `timestamps` 仅接受 true/false/'ms'/'s';单位换算在 Host(core 无时钟,now 由 Host 传入)
680fn parse_timestamps(obj: &Map<String, Value>) -> Result<bool, String> {
681    match obj.get("timestamps") {
682        None | Some(Value::Null) | Some(Value::Bool(true)) => Ok(true),
683        Some(Value::Bool(false)) => Ok(false),
684        Some(Value::String(s)) if s == "ms" || s == "s" => Ok(true),
685        Some(Value::String(s)) => Err(format!(
686            "timestamps 仅支持 true/false/'ms'/'s',实际 {:?}",
687            s
688        )),
689        Some(other) => Err(format!(
690            "timestamps 仅支持 true/false/'ms'/'s',实际 {}",
691            other
692        )),
693    }
694}
695
696/// 自动补时间戳字段(`timestamps !== false` 时;已声明则不覆盖)
697fn add_timestamp_fields(fields: &mut HashMap<String, FieldDef>) {
698    for key in ["createdAt", "updatedAt"] {
699        fields.entry(key.to_string()).or_insert_with(|| FieldDef {
700            field_type: "number".to_string(),
701            required: false,
702            default: None,
703            read: None,
704            write: None,
705            fields: None,
706            strategy: None,
707        });
708    }
709}
710
711fn parse_relations(obj: &Map<String, Value>) -> HashMap<String, RelationDef> {
712    let mut relations = HashMap::new();
713    if let Some(Value::Object(rm)) = obj.get("relations") {
714        for (key, val) in rm {
715            let vo = val.as_object();
716            let get_str = |k: &str| -> Option<String> {
717                vo.and_then(|o| o.get(k))
718                    .and_then(|v| v.as_str())
719                    .filter(|s| !s.is_empty())
720                    .map(String::from)
721            };
722            relations.insert(
723                key.clone(),
724                RelationDef {
725                    model: get_str("model").unwrap_or_default(),
726                    rel_type: get_str("type").unwrap_or_else(|| "many".to_string()),
727                    local_field: get_str("localField").unwrap_or_else(|| "_id".to_string()),
728                    foreign_field: get_str("foreignField").unwrap_or_else(|| key.clone()),
729                    read: str_list(vo.and_then(|o| o.get("read"))),
730                },
731            );
732        }
733    }
734    relations
735}
736
737fn parse_computes(
738    obj: &Map<String, Value>,
739    relations: &HashMap<String, RelationDef>,
740) -> Result<Vec<(String, ComputeDef)>, String> {
741    let mut computes = Vec::new();
742    if let Some(Value::Object(cm)) = obj.get("computes") {
743        for (key, val) in cm {
744            let vo = val.as_object();
745            let get = |k: &str| vo.and_then(|o| o.get(k)).cloned();
746            let has_fn = get("fn").map(|v| is_truthy(&v)).unwrap_or(false);
747            let has_async_fn = get("asyncFn").map(|v| is_truthy(&v)).unwrap_or(false);
748            let agg = parse_agg(key, get("agg").as_ref(), relations, has_fn || has_async_fn)?;
749            computes.push((
750                key.clone(),
751                ComputeDef {
752                    comp_type: get("type")
753                        .and_then(|v| v.as_str().map(String::from))
754                        .filter(|s| !s.is_empty())
755                        .unwrap_or_else(|| "any".to_string()),
756                    has_fn,
757                    has_async_fn,
758                    fn_ref: get("fnRef")
759                        .and_then(|v| v.as_str().map(String::from))
760                        .filter(|s| !s.is_empty()),
761                    agg,
762                    depends: str_list(get("depends").as_ref()).unwrap_or_default(),
763                    read: str_list(get("read").as_ref()),
764                },
765            ));
766        }
767    }
768    Ok(computes)
769}
770
771/// 解析并校验归一聚合计算列(§9.2(2)):`{"$count":"lessons"}` / `{"$sum":"lessons.duration"}`。
772///
773/// - 算子白名单 `$count/$sum/$avg/$min/$max`,不在白名单 → Err(绝不静默);
774/// - 取值必须是**关系路径**:`$count` 仅关系整行计数(`lessons`,不接受字段),
775///   其余算子必须带字段(`lessons.duration`);
776/// - 二级关系路径(`orders.items.price`)首批不支持 → Err;
777/// - 与 `fn` / `asyncFn` 互斥。
778fn parse_agg(
779    key: &str,
780    raw: Option<&Value>,
781    relations: &HashMap<String, RelationDef>,
782    has_callback: bool,
783) -> Result<Option<Value>, String> {
784    let Some(v) = raw.filter(|v| is_truthy(v)) else {
785        return Ok(None);
786    };
787    if has_callback {
788        return Err(format!(
789            "计算列 \"{key}\" 不能同时声明 agg 与 fn/asyncFn(二选一)"
790        ));
791    }
792    let obj = v.as_object().ok_or_else(|| {
793        format!("计算列 \"{key}\" 的 agg 必须是对象,如 {{\"$count\": \"lessons\"}}")
794    })?;
795    let mut it = obj.iter();
796    let (op, arg) = match (it.next(), it.next()) {
797        (Some(kv), None) => kv,
798        _ => return Err(format!("计算列 \"{key}\" 的 agg 必须且只能声明一个算子键")),
799    };
800    if !AGG_OPS.contains(&op.as_str()) {
801        return Err(format!(
802            "计算列 \"{key}\" 的 agg 算子 {op} 不在白名单($count/$sum/$avg/$min/$max)"
803        ));
804    }
805    let path = arg
806        .as_str()
807        .filter(|s| !s.is_empty() && *s != "*")
808        .ok_or_else(|| {
809            format!("计算列 \"{key}\" 的 agg 取值必须是关系路径字符串(如 \"lessons\" / \"lessons.duration\")")
810        })?;
811    let (rel, field) = match path.split_once('.') {
812        Some((r, f)) => (r, Some(f)),
813        None => (path, None),
814    };
815    if !relations.contains_key(rel) {
816        return Err(format!(
817            "计算列 \"{key}\" 的 agg 引用的关系 \"{rel}\" 未在 schema 中定义"
818        ));
819    }
820    if field
821        .map(|f| f.is_empty() || f.contains('.'))
822        .unwrap_or(false)
823    {
824        return Err(format!(
825            "计算列 \"{key}\" 的 agg 关系路径 \"{path}\" 非法:仅支持单级「关系.字段」(二级路径首批不支持)"
826        ));
827    }
828    match op.as_str() {
829        "$count" if field.is_some() => Err(format!(
830            "计算列 \"{key}\" 的 $count 仅支持关系整行计数(如 \"lessons\"),按字段计数待 P5"
831        )),
832        "$count" => Ok(Some(v.clone())),
833        _ if field.is_none() => Err(format!(
834            "计算列 \"{key}\" 的 agg 算子 {op} 必须引用关系字段(如 \"lessons.duration\")"
835        )),
836        _ => Ok(Some(v.clone())),
837    }
838}
839
840/// 构造 `<Name>Deleted` 归档表 schema 定义(字段 = 原字段 + `deletedAt`)。
841/// 归档表与原表同落点 —— 落点不由定义文件携带,改由批次继承主链路(见 `register_batch`)。
842fn archive_defn(obj: &Map<String, Value>, name: &str, collection: &str) -> Value {
843    let mut arch_fields = obj
844        .get("fields")
845        .and_then(|v| v.as_object())
846        .cloned()
847        .unwrap_or_default();
848    arch_fields.insert("deletedAt".to_string(), json!({ "type": "number" }));
849    // 剔除 `_id` 的自增策略:归档表显式拷贝源行 `_id` 值(upsert-by-_id 幂等),
850    // 不走数据库自增;DDL 侧也据此保持归档表 `_id` 为普通主键列
851    if let Some(idf) = arch_fields.get_mut("_id") {
852        if let Some(o) = idf.as_object_mut() {
853            o.remove("strategy");
854        }
855    }
856    let arch = json!({
857        "name": format!("{}Deleted", name),
858        "collection": format!("{}_deleted", collection),
859        "idPrefix": "",
860        "_isArchive": true,
861        "fields": Value::Object(arch_fields),
862        "indexes": obj.get("indexes").cloned().unwrap_or_else(|| json!([])),
863    });
864    arch
865}
866
867#[cfg(test)]
868mod tests {
869    use super::*;
870
871    #[test]
872    fn write_link_policy_defaults_to_reject() {
873        let r = Registry::new();
874        assert_eq!(r.write_link_policy(), WriteLinkPolicy::Reject);
875        let mut r2 = Registry::new();
876        assert_eq!(r2.write_link_policy(), WriteLinkPolicy::Reject);
877        r2.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
878        assert_eq!(r2.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
879    }
880}