1use std::collections::HashMap;
4use std::sync::Arc;
5
6use serde_json::{json, Map, Value};
7
8use crate::types::{is_truthy, str_list, COMPUTE_AGG_OPS};
9
10use super::definition::{
11 normalize_fields, validate_vector_fields, ComputeDef, FieldDef, Location, RelationDef, Schema,
12};
13use super::load::LoadConfig;
14
15use crate::command::WriteLinkPolicy;
16use crate::error::{CoreError, ERR_POLICY_VIEW_READONLY};
17
18#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
24pub enum Profile {
25 #[default]
27 Standard,
28 Text2Query,
30}
31
32impl Profile {
33 pub fn as_str(&self) -> &'static str {
35 match self {
36 Profile::Standard => "standard",
37 Profile::Text2Query => "text2query",
38 }
39 }
40
41 pub fn from_str_or_err(s: &str) -> Result<Profile, String> {
43 match s {
44 "standard" => Ok(Profile::Standard),
45 "text2query" => Ok(Profile::Text2Query),
46 other => Err(format!("未知 profile: {other}(仅 standard / text2query)")),
47 }
48 }
49}
50
51#[derive(Debug, Clone)]
53struct Entry {
54 schema: Schema,
55 links: Vec<Location>,
57 version: u64,
59}
60
61pub(super) struct BatchClass {
63 pub order: Vec<String>,
65 pub primaries: HashMap<String, (Value, Location, String)>,
67 pub replicas: HashMap<String, Vec<(Value, Location)>>,
69}
70
71pub(super) enum BatchError {
73 Parse(String),
75 PrimaryDuplicate { name: String, indices: Vec<usize> },
77 LocationConflict(Box<LocationConflict>),
79}
80
81pub(super) fn classify_batch(items: &[(Value, Location)]) -> Result<BatchClass, BatchError> {
87 let mut order: Vec<String> = Vec::new();
88 let mut primaries: HashMap<String, (Value, Location, String)> = HashMap::new();
89 let mut primary_idx: HashMap<String, usize> = HashMap::new();
90 let mut replicas: HashMap<String, Vec<(Value, Location)>> = HashMap::new();
91
92 for (idx, (defn, loc)) in items.iter().enumerate() {
93 let obj = defn
94 .as_object()
95 .ok_or_else(|| BatchError::Parse("schema 定义必须是对象".to_string()))?;
96 let name = obj
97 .get("name")
98 .and_then(|v| v.as_str())
99 .ok_or_else(|| BatchError::Parse("schema 缺少 name".to_string()))?
100 .to_string();
101 if !order.iter().any(|n| n == &name) {
102 order.push(name.clone());
103 }
104
105 if obj.get("replica").map(is_truthy).unwrap_or(false) {
106 replicas
107 .entry(name)
108 .or_default()
109 .push((defn.clone(), loc.clone()));
110 continue;
111 }
112
113 let collection = obj
114 .get("collection")
115 .and_then(|v| v.as_str())
116 .filter(|s| !s.is_empty())
117 .map(String::from)
118 .unwrap_or_else(|| name.clone());
119 if let Some(first) = primary_idx.get(&name) {
120 return Err(BatchError::PrimaryDuplicate {
122 name,
123 indices: vec![*first, idx],
124 });
125 }
126 primary_idx.insert(name.clone(), idx);
127 primaries.insert(name, (defn.clone(), loc.clone(), collection));
128 }
129
130 let mut quads: Vec<(String, String, Location)> = Vec::new();
132 for name in &order {
133 if let Some((_, loc, coll)) = primaries.get(name) {
134 quads.push((name.clone(), coll.clone(), loc.clone()));
135 }
136 if let Some(rs) = replicas.get(name) {
137 for (defn, loc) in rs {
138 let coll = defn
139 .get("name")
140 .and_then(|v| v.as_str())
141 .unwrap_or(name)
142 .to_string();
143 quads.push((name.clone(), coll, loc.clone()));
144 }
145 }
146 }
147 if let Some(c) = detect_location_conflict(&quads) {
148 return Err(BatchError::LocationConflict(Box::new(c)));
149 }
150
151 Ok(BatchClass {
152 order,
153 primaries,
154 replicas,
155 })
156}
157
158pub(super) struct LocationConflict {
160 pub collection: String,
161 pub prev: String,
162 pub name: String,
163 pub source: String,
164 pub database: Option<String>,
165 pub schema: Option<String>,
166}
167
168pub(super) fn detect_location_conflict(
172 entries: &[(String, String, Location)],
173) -> Option<LocationConflict> {
174 let mut seen: HashMap<(String, Option<String>, Option<String>, String), String> =
175 HashMap::new();
176 for (name, collection, loc) in entries {
177 let key = (
178 loc.source.clone(),
179 loc.database.clone(),
180 loc.schema.clone(),
181 collection.clone(),
182 );
183 if let Some(prev) = seen.get(&key) {
184 if prev != name {
185 return Some(LocationConflict {
186 collection: collection.clone(),
187 prev: prev.clone(),
188 name: name.clone(),
189 source: loc.source.clone(),
190 database: loc.database.clone(),
191 schema: loc.schema.clone(),
192 });
193 }
194 } else {
195 seen.insert(key, name.clone());
196 }
197 }
198 None
199}
200
201fn batch_err_to_msg(e: BatchError) -> String {
203 match e {
204 BatchError::Parse(m) => m,
205 BatchError::PrimaryDuplicate { name, .. } => format!(
206 "同名主定义重复: {}(同一装载批次内 name 必须唯一,从定义请用 replica: true)",
207 name
208 ),
209 BatchError::LocationConflict(c) => format!(
210 "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
211 c.source, c.database, c.schema, c.collection, c.prev
212 ),
213 }
214}
215
216#[derive(Debug, Clone, Default)]
218pub struct Catalog {
219 entries: HashMap<String, Entry>,
220 order: Vec<String>,
221}
222
223#[derive(Debug, Clone, Default)]
225pub struct PolicyBundle {
226 pub(crate) require_context: bool,
230 pub(crate) profile: Profile,
232 pub(crate) rbac: Option<crate::rbac::RbacPolicy>,
234 pub(crate) role_rules: crate::permission::RoleRules,
236 pub(crate) meta_policy: crate::permission::MetaPolicy,
238 pub(crate) write_link_policy: WriteLinkPolicy,
240}
241
242#[derive(Debug, Clone)]
247pub struct Registry {
248 catalog: Arc<Catalog>,
249 policy: PolicyBundle,
250 is_base: bool,
252}
253
254impl Default for Registry {
256 fn default() -> Self {
257 Self {
258 catalog: Arc::new(Catalog::default()),
259 policy: PolicyBundle::default(),
260 is_base: true,
261 }
262 }
263}
264
265#[derive(Debug, Clone, Default)]
268pub struct PolicyOverrides {
269 require_context: Option<bool>,
270 profile: Option<Profile>,
271 rbac: Option<Option<crate::rbac::RbacPolicy>>,
272 role_rules: Option<crate::permission::RoleRules>,
273 meta_policy: Option<crate::permission::MetaPolicy>,
274 write_link_policy: Option<WriteLinkPolicy>,
275}
276
277impl PolicyOverrides {
278 pub fn from_value(v: &Value) -> Result<Self, CoreError> {
284 let obj = v
285 .as_object()
286 .ok_or_else(|| CoreError::Other("策略覆盖必须是 JSON 对象".to_string()))?;
287 let mut o = PolicyOverrides::default();
288 for (k, val) in obj {
289 match k.as_str() {
290 "profile" => {
291 let s = val
292 .as_str()
293 .ok_or_else(|| CoreError::Other("profile 必须是字符串".to_string()))?;
294 o.profile = Some(Profile::from_str_or_err(s).map_err(CoreError::Other)?);
295 }
296 "requireContext" => {
297 o.require_context = Some(val.as_bool().ok_or_else(|| {
298 CoreError::Other("requireContext 必须是布尔".to_string())
299 })?);
300 }
301 "rbac" => {
302 o.rbac = Some(match val {
304 Value::Null => None,
305 other => Some(
306 crate::rbac::RbacPolicy::from_json(other).map_err(CoreError::Other)?,
307 ),
308 });
309 }
310 "roleRules" => {
311 o.role_rules = Some(parse_role_rules(val)?);
312 }
313 "metaPolicy" => {
314 o.meta_policy = Some(parse_meta_policy(val)?);
315 }
316 "writeLinkPolicy" => {
317 let s = val.as_str().ok_or_else(|| {
318 CoreError::Other("writeLinkPolicy 必须是字符串".to_string())
319 })?;
320 o.write_link_policy = Some(match s {
321 "reject" => WriteLinkPolicy::Reject,
322 "primaryOnly" => WriteLinkPolicy::PrimaryOnly,
323 other => {
324 return Err(CoreError::Other(format!(
325 "未知 writeLinkPolicy: {other}(仅 reject / primaryOnly)"
326 )))
327 }
328 });
329 }
330 other => {
331 return Err(CoreError::Other(format!(
332 "{ERR_POLICY_VIEW_READONLY} 未知策略覆盖键 `{other}`"
333 )))
334 }
335 }
336 }
337 Ok(o)
338 }
339}
340
341fn parse_role_rules(v: &Value) -> Result<crate::permission::RoleRules, CoreError> {
343 let obj = v
344 .as_object()
345 .ok_or_else(|| CoreError::Other("roleRules 必须是 JSON 对象".to_string()))?;
346 let mut rr = crate::permission::RoleRules::default();
347 for (k, val) in obj {
348 match k.as_str() {
349 "exemptRoles" => rr.exempt_roles = str_list_strict(val, "roleRules.exemptRoles")?,
350 "denyWriteRoles" => {
351 rr.deny_write_roles = str_list_strict(val, "roleRules.denyWriteRoles")?
352 }
353 "unconfigured" => {
354 let s = val.as_str().ok_or_else(|| {
355 CoreError::Other("roleRules.unconfigured 必须是字符串".to_string())
356 })?;
357 rr.unconfigured = crate::permission::UnconfiguredPolicy::from_str_or_err(s)
358 .map_err(CoreError::Other)?;
359 }
360 other => {
361 return Err(CoreError::Other(format!(
362 "{ERR_POLICY_VIEW_READONLY} 未知 roleRules 键 `{other}`"
363 )))
364 }
365 }
366 }
367 Ok(rr)
368}
369
370fn parse_meta_policy(v: &Value) -> Result<crate::permission::MetaPolicy, CoreError> {
372 let obj = v
373 .as_object()
374 .ok_or_else(|| CoreError::Other("metaPolicy 必须是 JSON 对象".to_string()))?;
375 let mut mp = crate::permission::MetaPolicy::default();
376 for (k, val) in obj {
377 match k.as_str() {
378 "closed" => {
379 mp.closed = val
380 .as_bool()
381 .ok_or_else(|| CoreError::Other("metaPolicy.closed 必须是布尔".to_string()))?;
382 }
383 "roles" => mp.roles = str_list_strict(val, "metaPolicy.roles")?,
384 other => {
385 return Err(CoreError::Other(format!(
386 "{ERR_POLICY_VIEW_READONLY} 未知 metaPolicy 键 `{other}`"
387 )))
388 }
389 }
390 }
391 Ok(mp)
392}
393
394fn str_list_strict(v: &Value, key: &str) -> Result<Vec<String>, CoreError> {
396 let arr = v
397 .as_array()
398 .ok_or_else(|| CoreError::Other(format!("{key} 必须是字符串数组")))?;
399 arr.iter()
400 .map(|x| {
401 x.as_str()
402 .map(String::from)
403 .ok_or_else(|| CoreError::Other(format!("{key} 的元素必须都是字符串")))
404 })
405 .collect()
406}
407
408impl PolicyBundle {
409 fn overridden(&self, o: &PolicyOverrides) -> PolicyBundle {
411 PolicyBundle {
412 require_context: o.require_context.unwrap_or(self.require_context),
413 profile: o.profile.unwrap_or(self.profile),
414 rbac: match &o.rbac {
415 Some(v) => v.clone(),
416 None => self.rbac.clone(),
417 },
418 role_rules: o
419 .role_rules
420 .clone()
421 .unwrap_or_else(|| self.role_rules.clone()),
422 meta_policy: o
423 .meta_policy
424 .clone()
425 .unwrap_or_else(|| self.meta_policy.clone()),
426 write_link_policy: o.write_link_policy.unwrap_or(self.write_link_policy),
427 }
428 }
429}
430
431#[derive(Debug, Clone)]
433pub struct ReadLinks {
434 pub links: Vec<Location>,
436 pub degraded: Vec<Value>,
438}
439
440impl Registry {
441 pub fn new() -> Self {
442 Self::default()
443 }
444
445 pub fn with_policy(&self, overrides: &PolicyOverrides) -> Registry {
450 Registry {
451 catalog: Arc::clone(&self.catalog),
452 policy: self.policy.overridden(overrides),
453 is_base: false,
454 }
455 }
456
457 pub fn is_view(&self) -> bool {
459 !self.is_base
460 }
461
462 fn ensure_base_catalog(&self) -> Result<(), String> {
468 if !self.is_base {
469 return Err(format!(
470 "{ERR_POLICY_VIEW_READONLY} 非 base 策略视图禁止注册/清空 schema(目录唯一真源在 base Registry)"
471 ));
472 }
473 Ok(())
474 }
475
476 pub fn register(&mut self, defn: &Value) -> Result<(), String> {
482 self.register_with_ctx(defn, None)
483 }
484
485 pub fn register_with_ctx(
487 &mut self,
488 defn: &Value,
489 ctx: Option<&crate::permission::Context>,
490 ) -> Result<(), String> {
491 self.register_batch(&[(defn.clone(), Location::default())], ctx)
492 }
493
494 pub fn register_batch(
509 &mut self,
510 items: &[(Value, Location)],
511 ctx: Option<&crate::permission::Context>,
512 ) -> Result<(), String> {
513 self.ensure_base_catalog()?; let class = classify_batch(items).map_err(batch_err_to_msg)?;
516
517 for (defn, _loc) in items {
519 let name = defn.get("name").and_then(|v| v.as_str()).unwrap_or("");
520 if !crate::permission::can_register(&self.policy.meta_policy, ctx) {
521 return Err(format!("ERR_PERMISSION: 无权注册或覆盖定义 {name}"));
522 }
523 }
524
525 struct Planned {
527 name: String,
528 schema: Schema,
529 links: Vec<Location>,
530 }
531 let mut planned: Vec<Planned> = Vec::new();
532 for name in &class.order {
533 let rs: Vec<Location> = class
534 .replicas
535 .get(name)
536 .map(|v| v.iter().map(|(_, l)| l.clone()).collect())
537 .unwrap_or_default();
538 if let Some((raw, loc, collection)) = class.primaries.get(name) {
539 let obj = raw.as_object().unwrap();
540 let schema = build_schema(obj, name.clone(), collection.clone(), loc)?;
541 let is_archive = obj.get("_isArchive").map(is_truthy).unwrap_or(false);
542 let mut links = vec![loc.clone()];
543 links.extend(rs);
544 planned.push(Planned {
545 name: name.clone(),
546 schema: schema.clone(),
547 links: links.clone(),
548 });
549 let archive_on = obj.get("archive").map(is_truthy).unwrap_or(true);
553 if !is_archive && !name.ends_with("Deleted") && archive_on {
554 let aobj = archive_defn(obj, name, &schema.collection)
555 .as_object()
556 .cloned()
557 .unwrap();
558 let aname = format!("{}Deleted", name);
559 let acoll = format!("{}_deleted", schema.collection);
560 let aschema = build_schema(&aobj, aname.clone(), acoll, &links[0])?;
561 planned.push(Planned {
562 name: aname,
563 schema: aschema,
564 links,
565 });
566 }
567 } else {
568 let Some(existing) = self.catalog.entries.get(name) else {
570 return Err(format!(
571 "未找到主 schema: {}(从定义 `replica` 必须伴随同批主定义,或该 name 已注册)",
572 name
573 ));
574 };
575 let mut links = existing.links.clone();
576 links.extend(rs);
577 planned.push(Planned {
578 name: name.clone(),
579 schema: existing.schema.clone(),
580 links,
581 });
582 }
583 }
584
585 let batch_names: std::collections::HashSet<&str> =
587 planned.iter().map(|p| p.name.as_str()).collect();
588 let mut quads: Vec<(String, String, Location)> = Vec::new();
589 for (n, e) in &self.catalog.entries {
590 if batch_names.contains(n.as_str()) {
591 continue; }
593 for l in &e.links {
594 quads.push((n.clone(), e.schema.collection.clone(), l.clone()));
595 }
596 }
597 for p in &planned {
598 for l in &p.links {
599 quads.push((p.name.clone(), p.schema.collection.clone(), l.clone()));
600 }
601 }
602 if let Some(c) = detect_location_conflict(&quads) {
603 return Err(format!(
604 "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
605 c.source, c.database, c.schema, c.collection, c.prev
606 ));
607 }
608
609 let catalog = Arc::make_mut(&mut self.catalog); for p in planned {
612 if let Some(e) = catalog.entries.get_mut(&p.name) {
613 e.schema = p.schema;
614 e.links = p.links;
615 e.version += 1;
616 } else {
617 catalog.entries.insert(
618 p.name.clone(),
619 Entry {
620 schema: p.schema,
621 links: p.links,
622 version: 1,
623 },
624 );
625 catalog.order.push(p.name);
626 }
627 }
628
629 Ok(())
630 }
631
632 pub fn apply_link_decls(&mut self, cfg: &LoadConfig) -> Result<(), String> {
641 if cfg.links.is_empty() {
642 return Ok(());
643 }
644 self.ensure_base_catalog()?; let declared: std::collections::HashSet<&str> =
648 cfg.links.iter().map(|(n, _)| n.as_str()).collect();
649 let mut planned: Vec<(String, Vec<Location>)> = Vec::new();
650 for (name, decls) in &cfg.links {
651 let entry = self.get_entry(name).map_err(|_| {
652 format!("未注册 schema: {name}(config.links 引用了不存在的 schema)")
653 })?;
654 let mut links = entry.links.clone();
655 for decl in decls {
656 let target = Location {
657 source: decl.source.clone(),
658 database: decl.database.clone(),
659 schema: decl.schema.clone(),
660 role: decl.role,
661 priority: decl.priority,
662 };
663 match links.iter_mut().find(|l| l.same_place(&target)) {
664 Some(hit) => {
665 hit.role = decl.role;
666 hit.priority = decl.priority;
667 }
668 None => links.push(target),
669 }
670 }
671 if !name.ends_with("Deleted") {
673 let aname = format!("{}Deleted", name);
674 if !declared.contains(aname.as_str()) && self.catalog.entries.contains_key(&aname) {
675 planned.push((aname, links.clone()));
676 }
677 }
678 planned.push((name.clone(), links));
679 }
680
681 let catalog = Arc::make_mut(&mut self.catalog);
683 for (name, links) in planned {
684 if let Some(e) = catalog.entries.get_mut(&name) {
685 e.links = links;
686 }
687 }
688 Ok(())
689 }
690
691 pub fn links_of(&self, name: &str) -> Result<&[Location], String> {
693 Ok(&self.get_entry(name)?.links)
694 }
695
696 pub fn resolve_read_links(
711 &self,
712 name: &str,
713 unreachable: Option<&[String]>,
714 ) -> Result<ReadLinks, String> {
715 let mut links = self.links_of(name)?.to_vec();
716 links.sort_by_key(|l| {
717 (
718 l.role.map_or(4u8, |r| r.rank()),
719 l.priority.unwrap_or(i64::MAX),
720 )
721 });
722
723 let mut degraded: Vec<Value> = Vec::new();
724 if let Some(unreachable) = unreachable {
725 links.retain(|l| {
726 if unreachable.iter().any(|s| s == &l.source) {
727 degraded.push(json!({
728 "code": "readLinkUnreachable",
729 "layer": "schema",
730 "message": format!("schema {} 的链路 {} 不可达,已跳过", name, l.source),
731 "hint": "检查该数据源连接;或调整 store.config.json 的 links 优先级",
732 "schema": name,
733 "source": l.source,
734 }));
735 false
736 } else {
737 true
738 }
739 });
740 }
741
742 if links.is_empty() {
743 return Err(format!(
744 "ERR_READ_NO_REACHABLE_LINK: {} 的全部链路不可达(已按 unreachable 跳过)",
745 name
746 ));
747 }
748 Ok(ReadLinks { links, degraded })
749 }
750
751 pub fn primary_location(&self, name: &str) -> Result<&Location, String> {
753 self.get_entry(name)?
754 .links
755 .first()
756 .ok_or_else(|| format!("Schema 无落点链路: {}", name))
757 }
758
759 pub fn version_of(&self, name: &str) -> Result<u64, String> {
761 Ok(self.get_entry(name)?.version)
762 }
763
764 fn get_entry(&self, name: &str) -> Result<&Entry, String> {
765 self.catalog
766 .entries
767 .get(name)
768 .ok_or_else(|| format!("Schema 未注册: {}", name))
769 }
770
771 pub fn get(&self, name: &str) -> Result<&Schema, String> {
773 self.get_entry(name).map(|e| &e.schema)
774 }
775
776 pub fn has(&self, name: &str) -> bool {
777 self.catalog.entries.contains_key(name)
778 }
779
780 pub fn clear(&mut self) {
789 if let Err(e) = self.ensure_base_catalog() {
790 panic!("{e}");
791 }
792 let catalog = Arc::make_mut(&mut self.catalog);
793 catalog.entries.clear();
794 catalog.order.clear();
795 }
796
797 pub fn set_require_context(&mut self, require: bool) {
801 self.policy.require_context = require;
802 }
803
804 pub fn require_context(&self) -> bool {
806 self.policy.require_context
807 }
808
809 pub fn set_profile(&mut self, profile: Profile) {
814 self.policy.profile = profile;
815 }
816
817 pub fn profile(&self) -> Profile {
819 self.policy.profile
820 }
821
822 pub fn set_exempt_roles(&mut self, roles: Vec<String>) {
824 self.policy.role_rules.exempt_roles = roles;
825 }
826
827 pub fn set_deny_write_roles(&mut self, roles: Vec<String>) {
829 self.policy.role_rules.deny_write_roles = roles;
830 }
831
832 pub fn set_unconfigured_policy(&mut self, policy: crate::permission::UnconfiguredPolicy) {
834 self.policy.role_rules.unconfigured = policy;
835 }
836
837 pub fn role_rules(&self) -> &crate::permission::RoleRules {
839 &self.policy.role_rules
840 }
841
842 pub fn set_meta_policy(&mut self, closed: bool, roles: Vec<String>) {
845 self.policy.meta_policy = crate::permission::MetaPolicy { closed, roles };
846 }
847
848 pub fn meta_policy(&self) -> &crate::permission::MetaPolicy {
850 &self.policy.meta_policy
851 }
852
853 pub fn set_write_link_policy(&mut self, policy: WriteLinkPolicy) {
858 self.policy.write_link_policy = policy;
859 }
860
861 pub fn write_link_policy(&self) -> WriteLinkPolicy {
863 self.policy.write_link_policy
864 }
865
866 pub fn can_register(&self, ctx: Option<&crate::permission::Context>) -> bool {
872 crate::permission::can_register(&self.policy.meta_policy, ctx)
873 }
874
875 pub fn set_rbac(&mut self, policy: Option<&Value>) -> Result<(), String> {
878 match policy {
879 None => {
880 self.policy.rbac = None;
881 Ok(())
882 }
883 Some(v) => {
884 let p = crate::rbac::RbacPolicy::from_json(v)?;
885 self.policy.rbac = Some(p);
886 Ok(())
887 }
888 }
889 }
890
891 pub fn rbac(&self) -> Option<&crate::rbac::RbacPolicy> {
893 self.policy.rbac.as_ref()
894 }
895
896 pub fn get_by_location(
901 &self,
902 source: &str,
903 database: Option<&str>,
904 schema: Option<&str>,
905 collection: &str,
906 ) -> Result<&Schema, String> {
907 let db = database.filter(|s| !s.is_empty());
908 let sc = schema.filter(|s| !s.is_empty());
909 self.catalog
910 .entries
911 .values()
912 .find(|e| {
913 e.schema.collection == collection
914 && e.links.iter().any(|l| {
915 l.source == source
916 && l.database.as_deref().filter(|s| !s.is_empty()) == db
917 && l.schema.as_deref().filter(|s| !s.is_empty()) == sc
918 })
919 })
920 .map(|e| &e.schema)
921 .ok_or_else(|| {
922 format!(
923 "Schema 未定位(source = {}, database = {:?}, schema = {:?}, collection = {})",
924 source, db, sc, collection
925 )
926 })
927 }
928
929 pub fn get_for_command(
936 &self,
937 source: &str,
938 database: Option<&str>,
939 schema: Option<&str>,
940 collection: &str,
941 ) -> Result<&Schema, String> {
942 if let Ok(s) = self.get_by_location(source, database, schema, collection) {
943 return Ok(s);
944 }
945 let candidates: Vec<&Entry> = self
946 .catalog
947 .entries
948 .values()
949 .filter(|e| {
950 e.schema.collection == collection && e.links.iter().any(|l| l.source == source)
951 })
952 .collect();
953 match candidates.len() {
954 0 => self.get_by_location(source, database, schema, collection),
955 1 => Ok(&candidates[0].schema),
956 _ => candidates
957 .iter()
958 .find(|e| e.schema.database().is_none() && e.schema.schema().is_none())
959 .map(|e| &e.schema)
960 .ok_or_else(|| {
961 format!(
962 "命令定位 ({}, {:?}, {:?}, {}) 存在多个结构 schema(多租户 override 需唯一的 (source, collection) 结构声明)",
963 source, database, schema, collection
964 )
965 }),
966 }
967 }
968
969 pub fn list(&self) -> Vec<String> {
970 self.catalog.order.clone()
971 }
972
973 pub fn schema_datasource(&self, name: &str) -> Result<Option<String>, String> {
975 Ok(self.get(name)?.source.clone())
976 }
977
978 pub fn resolve_datasource(
983 &self,
984 schema_name: &str,
985 config: &Value,
986 ) -> Result<crate::datasource::DataSource, String> {
987 let declared = self.get(schema_name)?.source.as_deref();
988 let cfg = crate::datasource::DataSourceConfig::from_json(config)?;
989 cfg.resolve(declared)
990 }
991}
992
993fn build_schema(
996 obj: &Map<String, Value>,
997 name: String,
998 collection: String,
999 loc: &Location,
1000) -> Result<Schema, String> {
1001 let timestamps = parse_timestamps(obj)?;
1002 let mut fields = normalize_fields(obj.get("fields"))?;
1003 validate_vector_fields(&fields)?;
1005 if timestamps {
1006 add_timestamp_fields(&mut fields);
1007 }
1008 let relations = parse_relations(obj);
1009 let computes = parse_computes(obj, &relations, &fields)?;
1011 let triggers = super::triggers::parse_triggers(obj, &name, &fields)?;
1013 for (k, ns) in fields
1018 .keys()
1019 .map(|k| (k, "fields"))
1020 .chain(relations.keys().map(|k| (k, "relations")))
1021 .chain(computes.iter().map(|(k, _)| (k, "computes")))
1022 {
1023 if k == "_score" {
1024 return Err(format!(
1025 "{}名称 \"_score\" 为向量检索评分保留键,不得在 schema {ns} 中声明",
1026 crate::naming::ERR_NAME_CONFLICT
1027 ));
1028 }
1029 }
1030 let mut ids: Vec<&str> = Vec::new();
1033 ids.extend(fields.keys().map(String::as_str));
1034 ids.extend(relations.keys().map(String::as_str));
1035 ids.extend(computes.iter().map(|(k, _)| k.as_str()));
1036 crate::naming::detect_conflicts(ids)?;
1037 let owner_key = match obj.get("ownerKey") {
1041 None | Some(Value::Null) => None,
1042 Some(Value::String(s)) if !s.is_empty() => Some(s.clone()),
1043 Some(other) => {
1044 return Err(format!(
1045 "schema \"{name}\" 的 ownerKey 必须为非空字符串,实际 {other}"
1046 ))
1047 }
1048 };
1049 match obj.get("archive") {
1053 None | Some(Value::Null) | Some(Value::Bool(_)) => {}
1054 Some(other) => {
1055 return Err(format!(
1056 "schema \"{name}\" 的 archive 必须为可选布尔值,实际 {other}"
1057 ))
1058 }
1059 }
1060 Ok(Schema {
1061 name,
1062 collection,
1063 id_prefix: obj
1064 .get("idPrefix")
1065 .and_then(|v| v.as_str())
1066 .unwrap_or("")
1067 .to_string(),
1068 timestamps,
1069 fields,
1070 relations,
1071 computes,
1072 triggers,
1073 read: str_list(obj.get("read")),
1074 write: str_list(obj.get("write")),
1075 indexes: obj
1076 .get("indexes")
1077 .and_then(|v| v.as_array())
1078 .cloned()
1079 .unwrap_or_default(),
1080 source: if loc.source == crate::datasource::DEFAULT_SOURCE {
1081 None
1082 } else {
1083 Some(loc.source.clone())
1084 },
1085 database: loc.database.clone().filter(|s| !s.is_empty()),
1086 schema: loc.schema.clone().filter(|s| !s.is_empty()),
1087 owner_key,
1088 })
1089}
1090
1091fn parse_timestamps(obj: &Map<String, Value>) -> Result<bool, String> {
1093 match obj.get("timestamps") {
1094 None | Some(Value::Null) | Some(Value::Bool(true)) => Ok(true),
1095 Some(Value::Bool(false)) => Ok(false),
1096 Some(Value::String(s)) if s == "ms" || s == "s" => Ok(true),
1097 Some(Value::String(s)) => Err(format!(
1098 "timestamps 仅支持 true/false/'ms'/'s',实际 {:?}",
1099 s
1100 )),
1101 Some(other) => Err(format!(
1102 "timestamps 仅支持 true/false/'ms'/'s',实际 {}",
1103 other
1104 )),
1105 }
1106}
1107
1108fn add_timestamp_fields(fields: &mut HashMap<String, FieldDef>) {
1110 for key in ["createdAt", "updatedAt"] {
1111 fields.entry(key.to_string()).or_insert_with(|| FieldDef {
1112 field_type: "number".to_string(),
1113 required: false,
1114 default: None,
1115 read: None,
1116 write: None,
1117 fields: None,
1118 strategy: None,
1119 dim: None,
1120 metric: None,
1121 });
1122 }
1123}
1124
1125fn parse_relations(obj: &Map<String, Value>) -> HashMap<String, RelationDef> {
1126 let mut relations = HashMap::new();
1127 if let Some(Value::Object(rm)) = obj.get("relations") {
1128 for (key, val) in rm {
1129 let vo = val.as_object();
1130 let get_str = |k: &str| -> Option<String> {
1131 vo.and_then(|o| o.get(k))
1132 .and_then(|v| v.as_str())
1133 .filter(|s| !s.is_empty())
1134 .map(String::from)
1135 };
1136 relations.insert(
1137 key.clone(),
1138 RelationDef {
1139 model: get_str("model").unwrap_or_default(),
1140 rel_type: get_str("type").unwrap_or_else(|| "many".to_string()),
1141 local_field: get_str("localField").unwrap_or_else(|| "_id".to_string()),
1142 foreign_field: get_str("foreignField").unwrap_or_else(|| key.clone()),
1143 read: str_list(vo.and_then(|o| o.get("read"))),
1144 },
1145 );
1146 }
1147 }
1148 relations
1149}
1150
1151fn parse_computes(
1152 obj: &Map<String, Value>,
1153 relations: &HashMap<String, RelationDef>,
1154 fields: &HashMap<String, FieldDef>,
1155) -> Result<Vec<(String, ComputeDef)>, String> {
1156 let mut computes = Vec::new();
1157 if let Some(Value::Object(cm)) = obj.get("computes") {
1158 for (key, val) in cm {
1159 let vo = val.as_object();
1160 let get = |k: &str| vo.and_then(|o| o.get(k)).cloned();
1161 let has_fn = get("fn").map(|v| is_truthy(&v)).unwrap_or(false);
1162 let has_async_fn = get("asyncFn").map(|v| is_truthy(&v)).unwrap_or(false);
1163 let agg = parse_agg(
1164 key,
1165 get("agg").as_ref(),
1166 relations,
1167 fields,
1168 str_list(get("depends").as_ref()).unwrap_or_default().as_slice(),
1169 has_fn || has_async_fn,
1170 )?;
1171 computes.push((
1172 key.clone(),
1173 ComputeDef {
1174 comp_type: get("type")
1175 .and_then(|v| v.as_str().map(String::from))
1176 .filter(|s| !s.is_empty())
1177 .unwrap_or_else(|| "any".to_string()),
1178 has_fn,
1179 has_async_fn,
1180 fn_ref: get("fnRef")
1181 .and_then(|v| v.as_str().map(String::from))
1182 .filter(|s| !s.is_empty()),
1183 agg,
1184 depends: str_list(get("depends").as_ref()).unwrap_or_default(),
1185 read: str_list(get("read").as_ref()),
1186 },
1187 ));
1188 }
1189 }
1190 Ok(computes)
1191}
1192
1193fn parse_agg(
1207 key: &str,
1208 raw: Option<&Value>,
1209 relations: &HashMap<String, RelationDef>,
1210 fields: &HashMap<String, FieldDef>,
1211 depends: &[String],
1212 has_callback: bool,
1213) -> Result<Option<Value>, String> {
1214 let Some(v) = raw.filter(|v| is_truthy(v)) else {
1215 return Ok(None);
1216 };
1217 if has_callback {
1218 return Err(format!(
1219 "计算列 \"{key}\" 不能同时声明 agg 与 fn/asyncFn(二选一)"
1220 ));
1221 }
1222 let obj = v.as_object().ok_or_else(|| {
1223 format!("计算列 \"{key}\" 的 agg 必须是对象,如 {{\"$count\": \"lessons\"}}")
1224 })?;
1225 let mut it = obj.iter();
1226 let (op, arg) = match (it.next(), it.next()) {
1227 (Some(kv), None) => kv,
1228 _ => return Err(format!("计算列 \"{key}\" 的 agg 必须且只能声明一个算子键")),
1229 };
1230 if !COMPUTE_AGG_OPS.contains(&op.as_str()) {
1231 return Err(format!(
1232 "计算列 \"{key}\" 的 agg 算子 {op} 不在白名单($count/$size/$sum/$avg/$min/$max/$countDistinct)"
1233 ));
1234 }
1235 let path = arg
1236 .as_str()
1237 .filter(|s| !s.is_empty() && *s != "*")
1238 .ok_or_else(|| {
1239 format!("计算列 \"{key}\" 的 agg 取值必须是路径字符串(如 \"lessons\" / \"lessons.duration\" / \"bidders.amount\")")
1240 })?;
1241 let (rel, field) = match path.split_once('.') {
1242 Some((r, f)) => (r, Some(f)),
1243 None => (path, None),
1244 };
1245 if field
1246 .map(|f| f.is_empty() || f.contains('.'))
1247 .unwrap_or(false)
1248 {
1249 return Err(format!(
1250 "计算列 \"{key}\" 的 agg 路径 \"{path}\" 非法:仅支持「关系.字段」/「数组.元素字段」单级路径(深层路径首批不支持)"
1251 ));
1252 }
1253 if relations.contains_key(rel) {
1254 match op.as_str() {
1256 "$count" if field.is_some() => Err(format!(
1257 "计算列 \"{key}\" 的 $count 仅支持关系整行计数(如 \"lessons\"),按字段计数待 P5"
1258 )),
1259 "$count" => Ok(Some(v.clone())),
1260 _ if field.is_none() => Err(format!(
1261 "计算列 \"{key}\" 的 agg 算子 {op} 必须引用关系字段(如 \"lessons.duration\")"
1262 )),
1263 _ => Ok(Some(v.clone())),
1264 }
1265 } else {
1266 match fields.get(rel) {
1267 Some(f) if f.field_type == "array" => {
1268 if !depends.iter().any(|d| d == rel) {
1270 return Err(format!(
1271 "计算列 \"{key}\" 的 depends 必须包含 inline agg 路径首段 \"{rel}\"(宿主本地重算依据)"
1272 ));
1273 }
1274 match op.as_str() {
1275 "$size" if field.is_some() => Err(format!(
1276 "计算列 \"{key}\" 的 $size 仅支持数组本身(如 \"{rel}\"),不接受元素字段"
1277 )),
1278 "$size" => Ok(Some(v.clone())),
1279 "$count" => Err(format!(
1280 "计算列 \"{key}\" 的 $count 不适用于数组字段:文档内数组计数用 $size(如 {{\"$size\": \"{rel}\"}})"
1281 )),
1282 _ if field.is_none() => Err(format!(
1283 "计算列 \"{key}\" 的 agg 算子 {op} 必须引用数组元素字段(如 \"{rel}.amount\")"
1284 )),
1285 _ => Ok(Some(v.clone())),
1286 }
1287 }
1288 Some(f) => Err(format!(
1289 "计算列 \"{key}\" 的 agg 路径首段 \"{rel}\" 是字段但类型为 \"{}\"(非数组),inline agg 仅支持数组字段",
1290 f.field_type
1291 )),
1292 None => Err(format!(
1293 "计算列 \"{key}\" 的 agg 引用的路径首段 \"{rel}\" 既非已定义关系也非数组字段"
1294 )),
1295 }
1296 }
1297}
1298
1299fn archive_defn(obj: &Map<String, Value>, name: &str, collection: &str) -> Value {
1302 let mut arch_fields = obj
1303 .get("fields")
1304 .and_then(|v| v.as_object())
1305 .cloned()
1306 .unwrap_or_default();
1307 arch_fields.insert("deletedAt".to_string(), json!({ "type": "number" }));
1308 if let Some(idf) = arch_fields.get_mut("_id") {
1311 if let Some(o) = idf.as_object_mut() {
1312 o.remove("strategy");
1313 }
1314 }
1315 let arch = json!({
1316 "name": format!("{}Deleted", name),
1317 "collection": format!("{}_deleted", collection),
1318 "idPrefix": "",
1319 "_isArchive": true,
1320 "fields": Value::Object(arch_fields),
1321 "indexes": obj.get("indexes").cloned().unwrap_or_else(|| json!([])),
1322 });
1323 arch
1324}
1325
1326#[cfg(test)]
1327mod tests {
1328 use super::*;
1329
1330 #[test]
1331 fn write_link_policy_defaults_to_reject() {
1332 let r = Registry::new();
1333 assert_eq!(r.write_link_policy(), WriteLinkPolicy::Reject);
1334 let mut r2 = Registry::new();
1335 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::Reject);
1336 r2.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
1337 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
1338 }
1339
1340 use crate::permission::UnconfiguredPolicy;
1343
1344 fn overrides_all() -> PolicyOverrides {
1346 PolicyOverrides::from_value(&json!({
1347 "profile": "standard",
1348 "requireContext": false,
1349 "rbac": null,
1350 "roleRules": { "exemptRoles": ["x"], "denyWriteRoles": ["y"], "unconfigured": "open" },
1351 "metaPolicy": { "closed": false, "roles": [] },
1352 "writeLinkPolicy": "reject"
1353 }))
1354 .expect("合法覆盖 JSON")
1355 }
1356
1357 #[test]
1359 fn with_policy_override_matrix() {
1360 let mut base = Registry::new();
1361 base.set_require_context(true);
1362 base.set_profile(Profile::Text2Query);
1363 base.set_rbac(Some(&json!({}))).expect("合法 RBAC");
1364 base.set_exempt_roles(vec!["a".to_string()]);
1365 base.set_deny_write_roles(vec!["b".to_string()]);
1366 base.set_unconfigured_policy(UnconfiguredPolicy::Closed);
1367 base.set_meta_policy(true, vec!["m".to_string()]);
1368 base.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
1369
1370 let inherit = base.with_policy(&PolicyOverrides::default());
1372 assert!(inherit.is_view(), "派生视图 is_view 应为真");
1373 assert!(inherit.require_context());
1374 assert_eq!(inherit.profile(), Profile::Text2Query);
1375 assert!(inherit.rbac().is_some());
1376 assert_eq!(inherit.role_rules().exempt_roles, vec!["a".to_string()]);
1377 assert_eq!(inherit.role_rules().deny_write_roles, vec!["b".to_string()]);
1378 assert_eq!(
1379 inherit.role_rules().unconfigured,
1380 UnconfiguredPolicy::Closed
1381 );
1382 assert!(inherit.meta_policy().closed);
1383 assert_eq!(inherit.meta_policy().roles, vec!["m".to_string()]);
1384 assert_eq!(inherit.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
1385
1386 let full = base.with_policy(&overrides_all());
1388 assert!(!full.require_context());
1389 assert_eq!(full.profile(), Profile::Standard);
1390 assert!(full.rbac().is_none(), "rbac:null 应清除");
1391 assert_eq!(full.role_rules().exempt_roles, vec!["x".to_string()]);
1392 assert_eq!(full.role_rules().deny_write_roles, vec!["y".to_string()]);
1393 assert_eq!(full.role_rules().unconfigured, UnconfiguredPolicy::Open);
1394 assert!(!full.meta_policy().closed);
1395 assert_eq!(full.write_link_policy(), WriteLinkPolicy::Reject);
1396
1397 assert!(base.require_context());
1399 assert_eq!(base.profile(), Profile::Text2Query);
1400 assert!(base.rbac().is_some());
1401 assert!(base.meta_policy().closed);
1402 }
1403
1404 #[test]
1406 fn view_register_is_readonly_err() {
1407 let base = Registry::new();
1408 let mut view = base.with_policy(&PolicyOverrides::default());
1409 let err = view
1410 .register(&json!({ "name": "A", "fields": {} }))
1411 .expect_err("视图注册应 Err");
1412 assert!(
1413 err.starts_with(ERR_POLICY_VIEW_READONLY),
1414 "前缀可匹配: {err}"
1415 );
1416 let err2 = view
1417 .register_with_ctx(&json!({ "name": "A", "fields": {} }), None)
1418 .expect_err("视图注册应 Err");
1419 assert!(
1420 err2.starts_with(ERR_POLICY_VIEW_READONLY),
1421 "前缀可匹配: {err2}"
1422 );
1423 let err3 = view
1424 .register_batch(
1425 &[(json!({ "name": "A", "fields": {} }), Location::default())],
1426 None,
1427 )
1428 .expect_err("视图批量注册应 Err");
1429 assert!(
1430 err3.starts_with(ERR_POLICY_VIEW_READONLY),
1431 "前缀可匹配: {err3}"
1432 );
1433 }
1434
1435 #[test]
1437 #[should_panic(expected = "ERR_POLICY_VIEW_READONLY")]
1438 fn view_clear_panics_readonly() {
1439 let base = Registry::new();
1440 let mut view = base.with_policy(&PolicyOverrides::default());
1441 view.clear();
1442 }
1443
1444 #[test]
1446 fn view_sees_catalog_snapshot() {
1447 let mut base = Registry::new();
1448 base.register(&json!({ "name": "A", "fields": { "x": { "type": "number" } } }))
1449 .expect("base 注册成功");
1450 let view = base.with_policy(&PolicyOverrides::default());
1451 assert!(view.has("A"), "派生视图应见 base 已注册目录");
1452 assert_eq!(view.get("A").expect("视图可取").name, "A");
1453 assert_eq!(view.list(), base.list());
1454
1455 let mut base2 = Registry::new();
1457 base2
1458 .register(&json!({ "name": "A", "fields": {} }))
1459 .unwrap();
1460 let snap = base2.with_policy(&PolicyOverrides::default());
1461 base2
1462 .register(&json!({ "name": "B", "fields": {} }))
1463 .unwrap();
1464 assert!(snap.has("A"), "在途视图仍持其快照");
1465 let fresh = base2.with_policy(&PolicyOverrides::default());
1466 assert!(fresh.has("B"), "新派生视图应见 base 后续注册");
1467 }
1468
1469 #[test]
1471 fn view_policy_set_isolated() {
1472 let base = Registry::new();
1473 let mut view = base.with_policy(&PolicyOverrides::default());
1474 view.set_profile(Profile::Text2Query);
1475 assert_eq!(view.profile(), Profile::Text2Query);
1476 assert_eq!(base.profile(), Profile::Standard, "视图 set 不影响 base");
1477 let sibling = base.with_policy(&PolicyOverrides::default());
1478 assert_eq!(sibling.profile(), Profile::Standard, "不影响兄弟视图");
1479 }
1480
1481 #[test]
1483 fn from_value_unknown_key_and_rbac_null() {
1484 let e = PolicyOverrides::from_value(&json!({ "nope": 1 })).expect_err("未知键应 Err");
1485 assert!(e.to_string().starts_with(ERR_POLICY_VIEW_READONLY), "{e}");
1486 assert!(
1487 PolicyOverrides::from_value(&json!({ "profile": "bogus" })).is_err(),
1488 "非法 profile 应 Err(禁静默回落)"
1489 );
1490 assert!(
1491 PolicyOverrides::from_value(&json!({ "writeLinkPolicy": "bogus" })).is_err(),
1492 "非法 writeLinkPolicy 应 Err"
1493 );
1494 assert!(
1495 PolicyOverrides::from_value(&json!({})).is_ok(),
1496 "空对象合法"
1497 );
1498
1499 let mut base = Registry::new();
1500 base.set_rbac(Some(&json!({}))).unwrap();
1501 let cleared =
1502 base.with_policy(&PolicyOverrides::from_value(&json!({ "rbac": null })).unwrap());
1503 assert!(cleared.rbac().is_none(), "rbac:null 应清除");
1504 let kept = base.with_policy(&PolicyOverrides::from_value(&json!({})).unwrap());
1505 assert!(kept.rbac().is_some(), "缺键应继承");
1506 assert!(base.rbac().is_some(), "base 不受影响");
1507 }
1508
1509 fn agg_computes(obj: &Value) -> Result<Vec<(String, ComputeDef)>, String> {
1513 let o = obj.as_object().expect("obj");
1514 let rels = parse_relations(o);
1515 let fields = normalize_fields(o.get("fields"))?;
1516 parse_computes(o, &rels, &fields)
1517 }
1518
1519 #[test]
1521 fn inline_agg_all_ops_ok() {
1522 let def = json!({
1523 "name": "Auction", "collection": "auctions", "timestamps": false,
1524 "fields": { "bidders": { "type": "array" } },
1525 "relations": {},
1526 "computes": {
1527 "biddersCount": { "type": "int", "agg": { "$size": "bidders" }, "depends": ["bidders"] },
1528 "maxBid": { "type": "number", "agg": { "$max": "bidders.amount" }, "depends": ["bidders"] },
1529 "bidderCount": { "type": "int", "agg": { "$countDistinct": "bidders.bidderId" }, "depends": ["bidders"] },
1530 "sumBid": { "type": "number", "agg": { "$sum": "bidders.amount" }, "depends": ["bidders"] },
1531 "avgBid": { "type": "number", "agg": { "$avg": "bidders.amount" }, "depends": ["bidders"] },
1532 "minBid": { "type": "number", "agg": { "$min": "bidders.amount" }, "depends": ["bidders"] }
1533 }
1534 });
1535 let computes = agg_computes(&def).expect("六算子 inline agg 均合法");
1536 assert_eq!(computes.len(), 6);
1537 for (_, c) in &computes {
1538 assert!(c.agg.is_some(), "inline agg 应保留");
1539 }
1540 }
1541
1542 #[test]
1544 fn inline_agg_without_depends_head_is_err() {
1545 let def = json!({
1546 "name": "A", "collection": "a", "timestamps": false,
1547 "fields": { "tags": { "type": "array" } },
1548 "relations": {},
1549 "computes": { "n": { "type": "int", "agg": { "$size": "tags" } } }
1550 });
1551 let err = agg_computes(&def).expect_err("缺 depends 应 Err");
1552 assert!(err.contains("depends"), "{err}");
1553 }
1554
1555 #[test]
1557 fn inline_agg_size_with_element_field_is_err() {
1558 let def = json!({
1559 "name": "A", "collection": "a", "timestamps": false,
1560 "fields": { "tags": { "type": "array" } },
1561 "relations": {},
1562 "computes": { "n": { "type": "int", "agg": { "$size": "tags.x" }, "depends": ["tags"] } }
1563 });
1564 let err = agg_computes(&def).expect_err("$size 带元素字段应 Err");
1565 assert!(err.contains("仅支持数组本身"), "{err}");
1566 }
1567
1568 #[test]
1570 fn inline_agg_count_is_err() {
1571 let def = json!({
1572 "name": "A", "collection": "a", "timestamps": false,
1573 "fields": { "tags": { "type": "array" } },
1574 "relations": {},
1575 "computes": { "n": { "type": "int", "agg": { "$count": "tags" }, "depends": ["tags"] } }
1576 });
1577 let err = agg_computes(&def).expect_err("数组 $count 应 Err");
1578 assert!(err.contains("$size"), "{err}");
1579 }
1580
1581 #[test]
1583 fn inline_agg_element_op_without_field_is_err() {
1584 let def = json!({
1585 "name": "A", "collection": "a", "timestamps": false,
1586 "fields": { "tags": { "type": "array" } },
1587 "relations": {},
1588 "computes": { "n": { "type": "int", "agg": { "$max": "tags" }, "depends": ["tags"] } }
1589 });
1590 let err = agg_computes(&def).expect_err("$max 无元素字段应 Err");
1591 assert!(err.contains("元素字段"), "{err}");
1592 }
1593
1594 #[test]
1596 fn agg_head_non_array_field_is_err() {
1597 let def = json!({
1598 "name": "A", "collection": "a", "timestamps": false,
1599 "fields": { "state": { "type": "string" } },
1600 "relations": {},
1601 "computes": { "n": { "type": "int", "agg": { "$size": "state" }, "depends": ["state"] } }
1602 });
1603 let err = agg_computes(&def).expect_err("非数组字段应 Err");
1604 assert!(err.contains("非数组"), "{err}");
1605 }
1606
1607 #[test]
1609 fn agg_head_unknown_is_err() {
1610 let def = json!({
1611 "name": "A", "collection": "a", "timestamps": false,
1612 "fields": { "state": { "type": "string" } },
1613 "relations": {},
1614 "computes": { "n": { "type": "int", "agg": { "$size": "ghost" }, "depends": ["ghost"] } }
1615 });
1616 let err = agg_computes(&def).expect_err("未知首段应 Err");
1617 assert!(err.contains("既非已定义关系也非数组字段"), "{err}");
1618 }
1619
1620 #[test]
1623 fn relation_agg_unchanged_by_inline_extension() {
1624 let def = json!({
1625 "name": "Shop", "collection": "shops", "timestamps": false,
1626 "fields": { "name": { "type": "string" } },
1627 "relations": { "orders": { "model": "Order", "type": "many", "localField": "_id", "foreignField": "shopId" } },
1628 "computes": {
1629 "orderCount": { "type": "int", "agg": { "$count": "orders" } },
1630 "totalAmount": { "type": "number", "agg": { "$sum": "orders.amount" } }
1631 }
1632 });
1633 let computes = agg_computes(&def).expect("relation agg 不受 04 扩展影响");
1634 assert_eq!(computes.len(), 2);
1635
1636 let bad = json!({
1637 "name": "Shop", "collection": "shops", "timestamps": false,
1638 "fields": { "name": { "type": "string" } },
1639 "relations": { "orders": { "model": "Order", "type": "many", "localField": "_id", "foreignField": "shopId" } },
1640 "computes": { "n": { "type": "int", "agg": { "$count": "orders.orderId" } } }
1641 });
1642 let err = agg_computes(&bad).expect_err("relation $count 带字段仍应 Err");
1643 assert!(err.contains("整行计数"), "{err}");
1644 }
1645}