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();
1035 ids.extend(fields.keys().map(String::as_str));
1036 ids.extend(relations.keys().map(String::as_str));
1037 ids.extend(computes.iter().map(|(k, _)| k.as_str()));
1038 crate::naming::detect_conflicts(ids)?;
1039 let owner_key = match obj.get("ownerKey") {
1043 None | Some(Value::Null) => None,
1044 Some(Value::String(s)) if !s.is_empty() => Some(s.clone()),
1045 Some(other) => {
1046 return Err(format!(
1047 "schema \"{name}\" 的 ownerKey 必须为非空字符串,实际 {other}"
1048 ))
1049 }
1050 };
1051 match obj.get("archive") {
1055 None | Some(Value::Null) | Some(Value::Bool(_)) => {}
1056 Some(other) => {
1057 return Err(format!(
1058 "schema \"{name}\" 的 archive 必须为可选布尔值,实际 {other}"
1059 ))
1060 }
1061 }
1062 Ok(Schema {
1063 name,
1064 collection,
1065 id_prefix: obj
1066 .get("idPrefix")
1067 .and_then(|v| v.as_str())
1068 .unwrap_or("")
1069 .to_string(),
1070 timestamps,
1071 fields,
1072 relations,
1073 computes,
1074 triggers,
1075 read: str_list(obj.get("read")),
1076 write: str_list(obj.get("write")),
1077 indexes: obj
1078 .get("indexes")
1079 .and_then(|v| v.as_array())
1080 .cloned()
1081 .unwrap_or_default(),
1082 source: if loc.source == crate::datasource::DEFAULT_SOURCE {
1083 None
1084 } else {
1085 Some(loc.source.clone())
1086 },
1087 database: loc.database.clone().filter(|s| !s.is_empty()),
1088 schema: loc.schema.clone().filter(|s| !s.is_empty()),
1089 owner_key,
1090 })
1091}
1092
1093fn parse_timestamps(obj: &Map<String, Value>) -> Result<bool, String> {
1095 match obj.get("timestamps") {
1096 None | Some(Value::Null) | Some(Value::Bool(true)) => Ok(true),
1097 Some(Value::Bool(false)) => Ok(false),
1098 Some(Value::String(s)) if s == "ms" || s == "s" => Ok(true),
1099 Some(Value::String(s)) => Err(format!(
1100 "timestamps 仅支持 true/false/'ms'/'s',实际 {:?}",
1101 s
1102 )),
1103 Some(other) => Err(format!(
1104 "timestamps 仅支持 true/false/'ms'/'s',实际 {}",
1105 other
1106 )),
1107 }
1108}
1109
1110fn add_timestamp_fields(fields: &mut HashMap<String, FieldDef>) {
1112 for key in ["createdAt", "updatedAt"] {
1113 fields.entry(key.to_string()).or_insert_with(|| FieldDef {
1114 field_type: "number".to_string(),
1115 required: false,
1116 default: None,
1117 read: None,
1118 write: None,
1119 fields: None,
1120 strategy: None,
1121 dim: None,
1122 metric: None,
1123 });
1124 }
1125}
1126
1127fn parse_relations(obj: &Map<String, Value>) -> HashMap<String, RelationDef> {
1128 let mut relations = HashMap::new();
1129 if let Some(Value::Object(rm)) = obj.get("relations") {
1130 for (key, val) in rm {
1131 let vo = val.as_object();
1132 let get_str = |k: &str| -> Option<String> {
1133 vo.and_then(|o| o.get(k))
1134 .and_then(|v| v.as_str())
1135 .filter(|s| !s.is_empty())
1136 .map(String::from)
1137 };
1138 relations.insert(
1139 key.clone(),
1140 RelationDef {
1141 model: get_str("model").unwrap_or_default(),
1142 rel_type: get_str("type").unwrap_or_else(|| "many".to_string()),
1143 local_field: get_str("localField").unwrap_or_else(|| "_id".to_string()),
1144 foreign_field: get_str("foreignField").unwrap_or_else(|| key.clone()),
1145 read: str_list(vo.and_then(|o| o.get("read"))),
1146 },
1147 );
1148 }
1149 }
1150 relations
1151}
1152
1153fn parse_computes(
1154 obj: &Map<String, Value>,
1155 relations: &HashMap<String, RelationDef>,
1156 fields: &HashMap<String, FieldDef>,
1157) -> Result<Vec<(String, ComputeDef)>, String> {
1158 let mut computes = Vec::new();
1159 if let Some(Value::Object(cm)) = obj.get("computes") {
1160 for (key, val) in cm {
1161 let vo = val.as_object();
1162 let get = |k: &str| vo.and_then(|o| o.get(k)).cloned();
1163 let has_fn = get("fn").map(|v| is_truthy(&v)).unwrap_or(false);
1164 let has_async_fn = get("asyncFn").map(|v| is_truthy(&v)).unwrap_or(false);
1165 let agg = parse_agg(
1166 key,
1167 get("agg").as_ref(),
1168 relations,
1169 fields,
1170 str_list(get("depends").as_ref())
1171 .unwrap_or_default()
1172 .as_slice(),
1173 has_fn || has_async_fn,
1174 )?;
1175 computes.push((
1176 key.clone(),
1177 ComputeDef {
1178 comp_type: get("type")
1179 .and_then(|v| v.as_str().map(String::from))
1180 .filter(|s| !s.is_empty())
1181 .unwrap_or_else(|| "any".to_string()),
1182 has_fn,
1183 has_async_fn,
1184 fn_ref: get("fnRef")
1185 .and_then(|v| v.as_str().map(String::from))
1186 .filter(|s| !s.is_empty()),
1187 agg,
1188 depends: str_list(get("depends").as_ref()).unwrap_or_default(),
1189 read: str_list(get("read").as_ref()),
1190 },
1191 ));
1192 }
1193 }
1194 Ok(computes)
1195}
1196
1197fn parse_agg(
1211 key: &str,
1212 raw: Option<&Value>,
1213 relations: &HashMap<String, RelationDef>,
1214 fields: &HashMap<String, FieldDef>,
1215 depends: &[String],
1216 has_callback: bool,
1217) -> Result<Option<Value>, String> {
1218 let Some(v) = raw.filter(|v| is_truthy(v)) else {
1219 return Ok(None);
1220 };
1221 if has_callback {
1222 return Err(format!(
1223 "计算列 \"{key}\" 不能同时声明 agg 与 fn/asyncFn(二选一)"
1224 ));
1225 }
1226 let obj = v.as_object().ok_or_else(|| {
1227 format!("计算列 \"{key}\" 的 agg 必须是对象,如 {{\"$count\": \"lessons\"}}")
1228 })?;
1229 let mut it = obj.iter();
1230 let (op, arg) = match (it.next(), it.next()) {
1231 (Some(kv), None) => kv,
1232 _ => return Err(format!("计算列 \"{key}\" 的 agg 必须且只能声明一个算子键")),
1233 };
1234 if !COMPUTE_AGG_OPS.contains(&op.as_str()) {
1235 return Err(format!(
1236 "计算列 \"{key}\" 的 agg 算子 {op} 不在白名单($count/$size/$sum/$avg/$min/$max/$countDistinct)"
1237 ));
1238 }
1239 let path = arg
1240 .as_str()
1241 .filter(|s| !s.is_empty() && *s != "*")
1242 .ok_or_else(|| {
1243 format!("计算列 \"{key}\" 的 agg 取值必须是路径字符串(如 \"lessons\" / \"lessons.duration\" / \"bidders.amount\")")
1244 })?;
1245 let (rel, field) = match path.split_once('.') {
1246 Some((r, f)) => (r, Some(f)),
1247 None => (path, None),
1248 };
1249 if field
1250 .map(|f| f.is_empty() || f.contains('.'))
1251 .unwrap_or(false)
1252 {
1253 return Err(format!(
1254 "计算列 \"{key}\" 的 agg 路径 \"{path}\" 非法:仅支持「关系.字段」/「数组.元素字段」单级路径(深层路径首批不支持)"
1255 ));
1256 }
1257 if relations.contains_key(rel) {
1258 match op.as_str() {
1260 "$count" if field.is_some() => Err(format!(
1261 "计算列 \"{key}\" 的 $count 仅支持关系整行计数(如 \"lessons\"),按字段计数待 P5"
1262 )),
1263 "$count" => Ok(Some(v.clone())),
1264 _ if field.is_none() => Err(format!(
1265 "计算列 \"{key}\" 的 agg 算子 {op} 必须引用关系字段(如 \"lessons.duration\")"
1266 )),
1267 _ => Ok(Some(v.clone())),
1268 }
1269 } else {
1270 match fields.get(rel) {
1271 Some(f) if f.field_type == "array" => {
1272 if !depends.iter().any(|d| d == rel) {
1274 return Err(format!(
1275 "计算列 \"{key}\" 的 depends 必须包含 inline agg 路径首段 \"{rel}\"(宿主本地重算依据)"
1276 ));
1277 }
1278 match op.as_str() {
1279 "$size" if field.is_some() => Err(format!(
1280 "计算列 \"{key}\" 的 $size 仅支持数组本身(如 \"{rel}\"),不接受元素字段"
1281 )),
1282 "$size" => Ok(Some(v.clone())),
1283 "$count" => Err(format!(
1284 "计算列 \"{key}\" 的 $count 不适用于数组字段:文档内数组计数用 $size(如 {{\"$size\": \"{rel}\"}})"
1285 )),
1286 _ if field.is_none() => Err(format!(
1287 "计算列 \"{key}\" 的 agg 算子 {op} 必须引用数组元素字段(如 \"{rel}.amount\")"
1288 )),
1289 _ => Ok(Some(v.clone())),
1290 }
1291 }
1292 Some(f) => Err(format!(
1293 "计算列 \"{key}\" 的 agg 路径首段 \"{rel}\" 是字段但类型为 \"{}\"(非数组),inline agg 仅支持数组字段",
1294 f.field_type
1295 )),
1296 None => Err(format!(
1297 "计算列 \"{key}\" 的 agg 引用的路径首段 \"{rel}\" 既非已定义关系也非数组字段"
1298 )),
1299 }
1300 }
1301}
1302
1303fn archive_defn(obj: &Map<String, Value>, name: &str, collection: &str) -> Value {
1306 let mut arch_fields = obj
1307 .get("fields")
1308 .and_then(|v| v.as_object())
1309 .cloned()
1310 .unwrap_or_default();
1311 arch_fields.insert("deletedAt".to_string(), json!({ "type": "number" }));
1312 if let Some(idf) = arch_fields.get_mut("_id") {
1315 if let Some(o) = idf.as_object_mut() {
1316 o.remove("strategy");
1317 }
1318 }
1319 let arch = json!({
1320 "name": format!("{}Deleted", name),
1321 "collection": format!("{}_deleted", collection),
1322 "idPrefix": "",
1323 "_isArchive": true,
1324 "fields": Value::Object(arch_fields),
1325 "indexes": obj.get("indexes").cloned().unwrap_or_else(|| json!([])),
1326 "timestamps": obj.get("timestamps").cloned().unwrap_or(json!(true)),
1331 });
1332 arch
1333}
1334
1335#[cfg(test)]
1336mod tests {
1337 use super::*;
1338
1339 #[test]
1340 fn write_link_policy_defaults_to_reject() {
1341 let r = Registry::new();
1342 assert_eq!(r.write_link_policy(), WriteLinkPolicy::Reject);
1343 let mut r2 = Registry::new();
1344 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::Reject);
1345 r2.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
1346 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
1347 }
1348
1349 use crate::permission::UnconfiguredPolicy;
1352
1353 fn overrides_all() -> PolicyOverrides {
1355 PolicyOverrides::from_value(&json!({
1356 "profile": "standard",
1357 "requireContext": false,
1358 "rbac": null,
1359 "roleRules": { "exemptRoles": ["x"], "denyWriteRoles": ["y"], "unconfigured": "open" },
1360 "metaPolicy": { "closed": false, "roles": [] },
1361 "writeLinkPolicy": "reject"
1362 }))
1363 .expect("合法覆盖 JSON")
1364 }
1365
1366 #[test]
1368 fn with_policy_override_matrix() {
1369 let mut base = Registry::new();
1370 base.set_require_context(true);
1371 base.set_profile(Profile::Text2Query);
1372 base.set_rbac(Some(&json!({}))).expect("合法 RBAC");
1373 base.set_exempt_roles(vec!["a".to_string()]);
1374 base.set_deny_write_roles(vec!["b".to_string()]);
1375 base.set_unconfigured_policy(UnconfiguredPolicy::Closed);
1376 base.set_meta_policy(true, vec!["m".to_string()]);
1377 base.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
1378
1379 let inherit = base.with_policy(&PolicyOverrides::default());
1381 assert!(inherit.is_view(), "派生视图 is_view 应为真");
1382 assert!(inherit.require_context());
1383 assert_eq!(inherit.profile(), Profile::Text2Query);
1384 assert!(inherit.rbac().is_some());
1385 assert_eq!(inherit.role_rules().exempt_roles, vec!["a".to_string()]);
1386 assert_eq!(inherit.role_rules().deny_write_roles, vec!["b".to_string()]);
1387 assert_eq!(
1388 inherit.role_rules().unconfigured,
1389 UnconfiguredPolicy::Closed
1390 );
1391 assert!(inherit.meta_policy().closed);
1392 assert_eq!(inherit.meta_policy().roles, vec!["m".to_string()]);
1393 assert_eq!(inherit.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
1394
1395 let full = base.with_policy(&overrides_all());
1397 assert!(!full.require_context());
1398 assert_eq!(full.profile(), Profile::Standard);
1399 assert!(full.rbac().is_none(), "rbac:null 应清除");
1400 assert_eq!(full.role_rules().exempt_roles, vec!["x".to_string()]);
1401 assert_eq!(full.role_rules().deny_write_roles, vec!["y".to_string()]);
1402 assert_eq!(full.role_rules().unconfigured, UnconfiguredPolicy::Open);
1403 assert!(!full.meta_policy().closed);
1404 assert_eq!(full.write_link_policy(), WriteLinkPolicy::Reject);
1405
1406 assert!(base.require_context());
1408 assert_eq!(base.profile(), Profile::Text2Query);
1409 assert!(base.rbac().is_some());
1410 assert!(base.meta_policy().closed);
1411 }
1412
1413 #[test]
1415 fn view_register_is_readonly_err() {
1416 let base = Registry::new();
1417 let mut view = base.with_policy(&PolicyOverrides::default());
1418 let err = view
1419 .register(&json!({ "name": "A", "fields": {} }))
1420 .expect_err("视图注册应 Err");
1421 assert!(
1422 err.starts_with(ERR_POLICY_VIEW_READONLY),
1423 "前缀可匹配: {err}"
1424 );
1425 let err2 = view
1426 .register_with_ctx(&json!({ "name": "A", "fields": {} }), None)
1427 .expect_err("视图注册应 Err");
1428 assert!(
1429 err2.starts_with(ERR_POLICY_VIEW_READONLY),
1430 "前缀可匹配: {err2}"
1431 );
1432 let err3 = view
1433 .register_batch(
1434 &[(json!({ "name": "A", "fields": {} }), Location::default())],
1435 None,
1436 )
1437 .expect_err("视图批量注册应 Err");
1438 assert!(
1439 err3.starts_with(ERR_POLICY_VIEW_READONLY),
1440 "前缀可匹配: {err3}"
1441 );
1442 }
1443
1444 #[test]
1446 #[should_panic(expected = "ERR_POLICY_VIEW_READONLY")]
1447 fn view_clear_panics_readonly() {
1448 let base = Registry::new();
1449 let mut view = base.with_policy(&PolicyOverrides::default());
1450 view.clear();
1451 }
1452
1453 #[test]
1456 fn field_named_type_is_allowed() {
1457 let mut r = Registry::new();
1458 r.register(&json!({
1459 "name": "Msg", "collection": "msg", "timestamps": false,
1460 "fields": {
1461 "_id": { "type": "string" },
1462 "type": { "type": "string", "default": "text" },
1463 "content": { "type": "string" }
1464 },
1465 "relations": {}
1466 }))
1467 .expect("字段名 type 合法,注册应成功");
1468 let s = r.get("Msg").expect("可取");
1469 let f = s.fields.get("type").expect("type 字段存在");
1470 assert_eq!(f.field_type, "string");
1471 assert_eq!(f.default, Some(json!("text")));
1472 }
1473
1474 #[test]
1479 fn archive_derivation_inherits_timestamps() {
1480 let mut r = Registry::new();
1481 r.register_batch(
1482 &[(
1483 json!({
1484 "name": "Vector", "collection": "vector", "timestamps": false,
1485 "fields": {
1486 "_id": { "type": "string" },
1487 "created_at": { "type": "string" },
1488 "updated_at": { "type": "string" }
1489 }
1490 }),
1491 Location::default(),
1492 )],
1493 None,
1494 )
1495 .expect("timestamps:false + snake_case 时间戳列应注册成功(归档派生不撞名)");
1496 let s = r.get("Vector").expect("主表可取");
1498 assert!(!s.timestamps);
1499 assert!(s.fields.contains_key("created_at"));
1500 assert!(
1501 !s.fields.contains_key("createdAt"),
1502 "timestamps:false 不注入 createdAt"
1503 );
1504 let arch = r.get("VectorDeleted").expect("归档表自动派生");
1506 assert_eq!(arch.collection, "vector_deleted");
1507 assert!(!arch.timestamps, "归档表应继承主表 timestamps:false");
1508 assert!(arch.fields.contains_key("created_at"), "归档表保留主表字段");
1509 assert!(
1510 !arch.fields.contains_key("createdAt"),
1511 "归档表继承 timestamps:false,不注入 createdAt"
1512 );
1513 }
1514
1515 #[test]
1517 fn view_sees_catalog_snapshot() {
1518 let mut base = Registry::new();
1519 base.register(&json!({ "name": "A", "fields": { "x": { "type": "number" } } }))
1520 .expect("base 注册成功");
1521 let view = base.with_policy(&PolicyOverrides::default());
1522 assert!(view.has("A"), "派生视图应见 base 已注册目录");
1523 assert_eq!(view.get("A").expect("视图可取").name, "A");
1524 assert_eq!(view.list(), base.list());
1525
1526 let mut base2 = Registry::new();
1528 base2
1529 .register(&json!({ "name": "A", "fields": {} }))
1530 .unwrap();
1531 let snap = base2.with_policy(&PolicyOverrides::default());
1532 base2
1533 .register(&json!({ "name": "B", "fields": {} }))
1534 .unwrap();
1535 assert!(snap.has("A"), "在途视图仍持其快照");
1536 let fresh = base2.with_policy(&PolicyOverrides::default());
1537 assert!(fresh.has("B"), "新派生视图应见 base 后续注册");
1538 }
1539
1540 #[test]
1542 fn view_policy_set_isolated() {
1543 let base = Registry::new();
1544 let mut view = base.with_policy(&PolicyOverrides::default());
1545 view.set_profile(Profile::Text2Query);
1546 assert_eq!(view.profile(), Profile::Text2Query);
1547 assert_eq!(base.profile(), Profile::Standard, "视图 set 不影响 base");
1548 let sibling = base.with_policy(&PolicyOverrides::default());
1549 assert_eq!(sibling.profile(), Profile::Standard, "不影响兄弟视图");
1550 }
1551
1552 #[test]
1554 fn from_value_unknown_key_and_rbac_null() {
1555 let e = PolicyOverrides::from_value(&json!({ "nope": 1 })).expect_err("未知键应 Err");
1556 assert!(e.to_string().starts_with(ERR_POLICY_VIEW_READONLY), "{e}");
1557 assert!(
1558 PolicyOverrides::from_value(&json!({ "profile": "bogus" })).is_err(),
1559 "非法 profile 应 Err(禁静默回落)"
1560 );
1561 assert!(
1562 PolicyOverrides::from_value(&json!({ "writeLinkPolicy": "bogus" })).is_err(),
1563 "非法 writeLinkPolicy 应 Err"
1564 );
1565 assert!(
1566 PolicyOverrides::from_value(&json!({})).is_ok(),
1567 "空对象合法"
1568 );
1569
1570 let mut base = Registry::new();
1571 base.set_rbac(Some(&json!({}))).unwrap();
1572 let cleared =
1573 base.with_policy(&PolicyOverrides::from_value(&json!({ "rbac": null })).unwrap());
1574 assert!(cleared.rbac().is_none(), "rbac:null 应清除");
1575 let kept = base.with_policy(&PolicyOverrides::from_value(&json!({})).unwrap());
1576 assert!(kept.rbac().is_some(), "缺键应继承");
1577 assert!(base.rbac().is_some(), "base 不受影响");
1578 }
1579
1580 fn agg_computes(obj: &Value) -> Result<Vec<(String, ComputeDef)>, String> {
1584 let o = obj.as_object().expect("obj");
1585 let rels = parse_relations(o);
1586 let fields = normalize_fields(o.get("fields"))?;
1587 parse_computes(o, &rels, &fields)
1588 }
1589
1590 #[test]
1592 fn inline_agg_all_ops_ok() {
1593 let def = json!({
1594 "name": "Auction", "collection": "auctions", "timestamps": false,
1595 "fields": { "bidders": { "type": "array" } },
1596 "relations": {},
1597 "computes": {
1598 "biddersCount": { "type": "int", "agg": { "$size": "bidders" }, "depends": ["bidders"] },
1599 "maxBid": { "type": "number", "agg": { "$max": "bidders.amount" }, "depends": ["bidders"] },
1600 "bidderCount": { "type": "int", "agg": { "$countDistinct": "bidders.bidderId" }, "depends": ["bidders"] },
1601 "sumBid": { "type": "number", "agg": { "$sum": "bidders.amount" }, "depends": ["bidders"] },
1602 "avgBid": { "type": "number", "agg": { "$avg": "bidders.amount" }, "depends": ["bidders"] },
1603 "minBid": { "type": "number", "agg": { "$min": "bidders.amount" }, "depends": ["bidders"] }
1604 }
1605 });
1606 let computes = agg_computes(&def).expect("六算子 inline agg 均合法");
1607 assert_eq!(computes.len(), 6);
1608 for (_, c) in &computes {
1609 assert!(c.agg.is_some(), "inline agg 应保留");
1610 }
1611 }
1612
1613 #[test]
1615 fn inline_agg_without_depends_head_is_err() {
1616 let def = json!({
1617 "name": "A", "collection": "a", "timestamps": false,
1618 "fields": { "tags": { "type": "array" } },
1619 "relations": {},
1620 "computes": { "n": { "type": "int", "agg": { "$size": "tags" } } }
1621 });
1622 let err = agg_computes(&def).expect_err("缺 depends 应 Err");
1623 assert!(err.contains("depends"), "{err}");
1624 }
1625
1626 #[test]
1628 fn inline_agg_size_with_element_field_is_err() {
1629 let def = json!({
1630 "name": "A", "collection": "a", "timestamps": false,
1631 "fields": { "tags": { "type": "array" } },
1632 "relations": {},
1633 "computes": { "n": { "type": "int", "agg": { "$size": "tags.x" }, "depends": ["tags"] } }
1634 });
1635 let err = agg_computes(&def).expect_err("$size 带元素字段应 Err");
1636 assert!(err.contains("仅支持数组本身"), "{err}");
1637 }
1638
1639 #[test]
1641 fn inline_agg_count_is_err() {
1642 let def = json!({
1643 "name": "A", "collection": "a", "timestamps": false,
1644 "fields": { "tags": { "type": "array" } },
1645 "relations": {},
1646 "computes": { "n": { "type": "int", "agg": { "$count": "tags" }, "depends": ["tags"] } }
1647 });
1648 let err = agg_computes(&def).expect_err("数组 $count 应 Err");
1649 assert!(err.contains("$size"), "{err}");
1650 }
1651
1652 #[test]
1654 fn inline_agg_element_op_without_field_is_err() {
1655 let def = json!({
1656 "name": "A", "collection": "a", "timestamps": false,
1657 "fields": { "tags": { "type": "array" } },
1658 "relations": {},
1659 "computes": { "n": { "type": "int", "agg": { "$max": "tags" }, "depends": ["tags"] } }
1660 });
1661 let err = agg_computes(&def).expect_err("$max 无元素字段应 Err");
1662 assert!(err.contains("元素字段"), "{err}");
1663 }
1664
1665 #[test]
1667 fn agg_head_non_array_field_is_err() {
1668 let def = json!({
1669 "name": "A", "collection": "a", "timestamps": false,
1670 "fields": { "state": { "type": "string" } },
1671 "relations": {},
1672 "computes": { "n": { "type": "int", "agg": { "$size": "state" }, "depends": ["state"] } }
1673 });
1674 let err = agg_computes(&def).expect_err("非数组字段应 Err");
1675 assert!(err.contains("非数组"), "{err}");
1676 }
1677
1678 #[test]
1680 fn agg_head_unknown_is_err() {
1681 let def = json!({
1682 "name": "A", "collection": "a", "timestamps": false,
1683 "fields": { "state": { "type": "string" } },
1684 "relations": {},
1685 "computes": { "n": { "type": "int", "agg": { "$size": "ghost" }, "depends": ["ghost"] } }
1686 });
1687 let err = agg_computes(&def).expect_err("未知首段应 Err");
1688 assert!(err.contains("既非已定义关系也非数组字段"), "{err}");
1689 }
1690
1691 #[test]
1694 fn relation_agg_unchanged_by_inline_extension() {
1695 let def = json!({
1696 "name": "Shop", "collection": "shops", "timestamps": false,
1697 "fields": { "name": { "type": "string" } },
1698 "relations": { "orders": { "model": "Order", "type": "many", "localField": "_id", "foreignField": "shopId" } },
1699 "computes": {
1700 "orderCount": { "type": "int", "agg": { "$count": "orders" } },
1701 "totalAmount": { "type": "number", "agg": { "$sum": "orders.amount" } }
1702 }
1703 });
1704 let computes = agg_computes(&def).expect("relation agg 不受 04 扩展影响");
1705 assert_eq!(computes.len(), 2);
1706
1707 let bad = json!({
1708 "name": "Shop", "collection": "shops", "timestamps": false,
1709 "fields": { "name": { "type": "string" } },
1710 "relations": { "orders": { "model": "Order", "type": "many", "localField": "_id", "foreignField": "shopId" } },
1711 "computes": { "n": { "type": "int", "agg": { "$count": "orders.orderId" } } }
1712 });
1713 let err = agg_computes(&bad).expect_err("relation $count 带字段仍应 Err");
1714 assert!(err.contains("整行计数"), "{err}");
1715 }
1716}