1use std::collections::HashMap;
4use std::sync::Arc;
5
6use serde_json::{json, Map, Value};
7
8use crate::types::{is_truthy, str_list, AGG_OPS};
9
10use super::definition::{normalize_fields, ComputeDef, FieldDef, Location, RelationDef, Schema};
11use super::load::LoadConfig;
12
13use crate::command::WriteLinkPolicy;
14use crate::error::{CoreError, ERR_POLICY_VIEW_READONLY};
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
22pub enum Profile {
23 #[default]
25 Standard,
26 Text2Query,
28}
29
30impl Profile {
31 pub fn as_str(&self) -> &'static str {
33 match self {
34 Profile::Standard => "standard",
35 Profile::Text2Query => "text2query",
36 }
37 }
38
39 pub fn from_str_or_err(s: &str) -> Result<Profile, String> {
41 match s {
42 "standard" => Ok(Profile::Standard),
43 "text2query" => Ok(Profile::Text2Query),
44 other => Err(format!("未知 profile: {other}(仅 standard / text2query)")),
45 }
46 }
47}
48
49#[derive(Debug, Clone)]
51struct Entry {
52 schema: Schema,
53 links: Vec<Location>,
55 version: u64,
57}
58
59pub(super) struct BatchClass {
61 pub order: Vec<String>,
63 pub primaries: HashMap<String, (Value, Location, String)>,
65 pub replicas: HashMap<String, Vec<(Value, Location)>>,
67}
68
69pub(super) enum BatchError {
71 Parse(String),
73 PrimaryDuplicate { name: String, indices: Vec<usize> },
75 LocationConflict(Box<LocationConflict>),
77}
78
79pub(super) fn classify_batch(items: &[(Value, Location)]) -> Result<BatchClass, BatchError> {
85 let mut order: Vec<String> = Vec::new();
86 let mut primaries: HashMap<String, (Value, Location, String)> = HashMap::new();
87 let mut primary_idx: HashMap<String, usize> = HashMap::new();
88 let mut replicas: HashMap<String, Vec<(Value, Location)>> = HashMap::new();
89
90 for (idx, (defn, loc)) in items.iter().enumerate() {
91 let obj = defn
92 .as_object()
93 .ok_or_else(|| BatchError::Parse("schema 定义必须是对象".to_string()))?;
94 let name = obj
95 .get("name")
96 .and_then(|v| v.as_str())
97 .ok_or_else(|| BatchError::Parse("schema 缺少 name".to_string()))?
98 .to_string();
99 if !order.iter().any(|n| n == &name) {
100 order.push(name.clone());
101 }
102
103 if obj.get("replica").map(is_truthy).unwrap_or(false) {
104 replicas
105 .entry(name)
106 .or_default()
107 .push((defn.clone(), loc.clone()));
108 continue;
109 }
110
111 let collection = obj
112 .get("collection")
113 .and_then(|v| v.as_str())
114 .filter(|s| !s.is_empty())
115 .map(String::from)
116 .unwrap_or_else(|| name.clone());
117 if let Some(first) = primary_idx.get(&name) {
118 return Err(BatchError::PrimaryDuplicate {
120 name,
121 indices: vec![*first, idx],
122 });
123 }
124 primary_idx.insert(name.clone(), idx);
125 primaries.insert(name, (defn.clone(), loc.clone(), collection));
126 }
127
128 let mut quads: Vec<(String, String, Location)> = Vec::new();
130 for name in &order {
131 if let Some((_, loc, coll)) = primaries.get(name) {
132 quads.push((name.clone(), coll.clone(), loc.clone()));
133 }
134 if let Some(rs) = replicas.get(name) {
135 for (defn, loc) in rs {
136 let coll = defn
137 .get("name")
138 .and_then(|v| v.as_str())
139 .unwrap_or(name)
140 .to_string();
141 quads.push((name.clone(), coll, loc.clone()));
142 }
143 }
144 }
145 if let Some(c) = detect_location_conflict(&quads) {
146 return Err(BatchError::LocationConflict(Box::new(c)));
147 }
148
149 Ok(BatchClass {
150 order,
151 primaries,
152 replicas,
153 })
154}
155
156pub(super) struct LocationConflict {
158 pub collection: String,
159 pub prev: String,
160 pub name: String,
161 pub source: String,
162 pub database: Option<String>,
163 pub schema: Option<String>,
164}
165
166pub(super) fn detect_location_conflict(
170 entries: &[(String, String, Location)],
171) -> Option<LocationConflict> {
172 let mut seen: HashMap<(String, Option<String>, Option<String>, String), String> =
173 HashMap::new();
174 for (name, collection, loc) in entries {
175 let key = (
176 loc.source.clone(),
177 loc.database.clone(),
178 loc.schema.clone(),
179 collection.clone(),
180 );
181 if let Some(prev) = seen.get(&key) {
182 if prev != name {
183 return Some(LocationConflict {
184 collection: collection.clone(),
185 prev: prev.clone(),
186 name: name.clone(),
187 source: loc.source.clone(),
188 database: loc.database.clone(),
189 schema: loc.schema.clone(),
190 });
191 }
192 } else {
193 seen.insert(key, name.clone());
194 }
195 }
196 None
197}
198
199fn batch_err_to_msg(e: BatchError) -> String {
201 match e {
202 BatchError::Parse(m) => m,
203 BatchError::PrimaryDuplicate { name, .. } => format!(
204 "同名主定义重复: {}(同一装载批次内 name 必须唯一,从定义请用 replica: true)",
205 name
206 ),
207 BatchError::LocationConflict(c) => format!(
208 "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
209 c.source, c.database, c.schema, c.collection, c.prev
210 ),
211 }
212}
213
214#[derive(Debug, Clone, Default)]
216pub struct Catalog {
217 entries: HashMap<String, Entry>,
218 order: Vec<String>,
219}
220
221#[derive(Debug, Clone, Default)]
223pub struct PolicyBundle {
224 pub(crate) require_context: bool,
228 pub(crate) profile: Profile,
230 pub(crate) rbac: Option<crate::rbac::RbacPolicy>,
232 pub(crate) role_rules: crate::permission::RoleRules,
234 pub(crate) meta_policy: crate::permission::MetaPolicy,
236 pub(crate) write_link_policy: WriteLinkPolicy,
238}
239
240#[derive(Debug, Clone)]
245pub struct Registry {
246 catalog: Arc<Catalog>,
247 policy: PolicyBundle,
248 is_base: bool,
250}
251
252impl Default for Registry {
254 fn default() -> Self {
255 Self {
256 catalog: Arc::new(Catalog::default()),
257 policy: PolicyBundle::default(),
258 is_base: true,
259 }
260 }
261}
262
263#[derive(Debug, Clone, Default)]
266pub struct PolicyOverrides {
267 require_context: Option<bool>,
268 profile: Option<Profile>,
269 rbac: Option<Option<crate::rbac::RbacPolicy>>,
270 role_rules: Option<crate::permission::RoleRules>,
271 meta_policy: Option<crate::permission::MetaPolicy>,
272 write_link_policy: Option<WriteLinkPolicy>,
273}
274
275impl PolicyOverrides {
276 pub fn from_value(v: &Value) -> Result<Self, CoreError> {
282 let obj = v
283 .as_object()
284 .ok_or_else(|| CoreError::Other("策略覆盖必须是 JSON 对象".to_string()))?;
285 let mut o = PolicyOverrides::default();
286 for (k, val) in obj {
287 match k.as_str() {
288 "profile" => {
289 let s = val
290 .as_str()
291 .ok_or_else(|| CoreError::Other("profile 必须是字符串".to_string()))?;
292 o.profile = Some(Profile::from_str_or_err(s).map_err(CoreError::Other)?);
293 }
294 "requireContext" => {
295 o.require_context = Some(val.as_bool().ok_or_else(|| {
296 CoreError::Other("requireContext 必须是布尔".to_string())
297 })?);
298 }
299 "rbac" => {
300 o.rbac = Some(match val {
302 Value::Null => None,
303 other => Some(
304 crate::rbac::RbacPolicy::from_json(other).map_err(CoreError::Other)?,
305 ),
306 });
307 }
308 "roleRules" => {
309 o.role_rules = Some(parse_role_rules(val)?);
310 }
311 "metaPolicy" => {
312 o.meta_policy = Some(parse_meta_policy(val)?);
313 }
314 "writeLinkPolicy" => {
315 let s = val.as_str().ok_or_else(|| {
316 CoreError::Other("writeLinkPolicy 必须是字符串".to_string())
317 })?;
318 o.write_link_policy = Some(match s {
319 "reject" => WriteLinkPolicy::Reject,
320 "primaryOnly" => WriteLinkPolicy::PrimaryOnly,
321 other => {
322 return Err(CoreError::Other(format!(
323 "未知 writeLinkPolicy: {other}(仅 reject / primaryOnly)"
324 )))
325 }
326 });
327 }
328 other => {
329 return Err(CoreError::Other(format!(
330 "{ERR_POLICY_VIEW_READONLY} 未知策略覆盖键 `{other}`"
331 )))
332 }
333 }
334 }
335 Ok(o)
336 }
337}
338
339fn parse_role_rules(v: &Value) -> Result<crate::permission::RoleRules, CoreError> {
341 let obj = v
342 .as_object()
343 .ok_or_else(|| CoreError::Other("roleRules 必须是 JSON 对象".to_string()))?;
344 let mut rr = crate::permission::RoleRules::default();
345 for (k, val) in obj {
346 match k.as_str() {
347 "exemptRoles" => rr.exempt_roles = str_list_strict(val, "roleRules.exemptRoles")?,
348 "denyWriteRoles" => {
349 rr.deny_write_roles = str_list_strict(val, "roleRules.denyWriteRoles")?
350 }
351 "unconfigured" => {
352 let s = val.as_str().ok_or_else(|| {
353 CoreError::Other("roleRules.unconfigured 必须是字符串".to_string())
354 })?;
355 rr.unconfigured = crate::permission::UnconfiguredPolicy::from_str_or_err(s)
356 .map_err(CoreError::Other)?;
357 }
358 other => {
359 return Err(CoreError::Other(format!(
360 "{ERR_POLICY_VIEW_READONLY} 未知 roleRules 键 `{other}`"
361 )))
362 }
363 }
364 }
365 Ok(rr)
366}
367
368fn parse_meta_policy(v: &Value) -> Result<crate::permission::MetaPolicy, CoreError> {
370 let obj = v
371 .as_object()
372 .ok_or_else(|| CoreError::Other("metaPolicy 必须是 JSON 对象".to_string()))?;
373 let mut mp = crate::permission::MetaPolicy::default();
374 for (k, val) in obj {
375 match k.as_str() {
376 "closed" => {
377 mp.closed = val
378 .as_bool()
379 .ok_or_else(|| CoreError::Other("metaPolicy.closed 必须是布尔".to_string()))?;
380 }
381 "roles" => mp.roles = str_list_strict(val, "metaPolicy.roles")?,
382 other => {
383 return Err(CoreError::Other(format!(
384 "{ERR_POLICY_VIEW_READONLY} 未知 metaPolicy 键 `{other}`"
385 )))
386 }
387 }
388 }
389 Ok(mp)
390}
391
392fn str_list_strict(v: &Value, key: &str) -> Result<Vec<String>, CoreError> {
394 let arr = v
395 .as_array()
396 .ok_or_else(|| CoreError::Other(format!("{key} 必须是字符串数组")))?;
397 arr.iter()
398 .map(|x| {
399 x.as_str()
400 .map(String::from)
401 .ok_or_else(|| CoreError::Other(format!("{key} 的元素必须都是字符串")))
402 })
403 .collect()
404}
405
406impl PolicyBundle {
407 fn overridden(&self, o: &PolicyOverrides) -> PolicyBundle {
409 PolicyBundle {
410 require_context: o.require_context.unwrap_or(self.require_context),
411 profile: o.profile.unwrap_or(self.profile),
412 rbac: match &o.rbac {
413 Some(v) => v.clone(),
414 None => self.rbac.clone(),
415 },
416 role_rules: o
417 .role_rules
418 .clone()
419 .unwrap_or_else(|| self.role_rules.clone()),
420 meta_policy: o
421 .meta_policy
422 .clone()
423 .unwrap_or_else(|| self.meta_policy.clone()),
424 write_link_policy: o.write_link_policy.unwrap_or(self.write_link_policy),
425 }
426 }
427}
428
429#[derive(Debug, Clone)]
431pub struct ReadLinks {
432 pub links: Vec<Location>,
434 pub degraded: Vec<Value>,
436}
437
438impl Registry {
439 pub fn new() -> Self {
440 Self::default()
441 }
442
443 pub fn with_policy(&self, overrides: &PolicyOverrides) -> Registry {
448 Registry {
449 catalog: Arc::clone(&self.catalog),
450 policy: self.policy.overridden(overrides),
451 is_base: false,
452 }
453 }
454
455 pub fn is_view(&self) -> bool {
457 !self.is_base
458 }
459
460 fn ensure_base_catalog(&self) -> Result<(), String> {
466 if !self.is_base {
467 return Err(format!(
468 "{ERR_POLICY_VIEW_READONLY} 非 base 策略视图禁止注册/清空 schema(目录唯一真源在 base Registry)"
469 ));
470 }
471 Ok(())
472 }
473
474 pub fn register(&mut self, defn: &Value) -> Result<(), String> {
480 self.register_with_ctx(defn, None)
481 }
482
483 pub fn register_with_ctx(
485 &mut self,
486 defn: &Value,
487 ctx: Option<&crate::permission::Context>,
488 ) -> Result<(), String> {
489 self.register_batch(&[(defn.clone(), Location::default())], ctx)
490 }
491
492 pub fn register_batch(
507 &mut self,
508 items: &[(Value, Location)],
509 ctx: Option<&crate::permission::Context>,
510 ) -> Result<(), String> {
511 self.ensure_base_catalog()?; let class = classify_batch(items).map_err(batch_err_to_msg)?;
514
515 for (defn, _loc) in items {
517 let name = defn.get("name").and_then(|v| v.as_str()).unwrap_or("");
518 if !crate::permission::can_register(&self.policy.meta_policy, ctx) {
519 return Err(format!("ERR_PERMISSION: 无权注册或覆盖定义 {name}"));
520 }
521 }
522
523 struct Planned {
525 name: String,
526 schema: Schema,
527 links: Vec<Location>,
528 }
529 let mut planned: Vec<Planned> = Vec::new();
530 for name in &class.order {
531 let rs: Vec<Location> = class
532 .replicas
533 .get(name)
534 .map(|v| v.iter().map(|(_, l)| l.clone()).collect())
535 .unwrap_or_default();
536 if let Some((raw, loc, collection)) = class.primaries.get(name) {
537 let obj = raw.as_object().unwrap();
538 let schema = build_schema(obj, name.clone(), collection.clone(), loc)?;
539 let is_archive = obj.get("_isArchive").map(is_truthy).unwrap_or(false);
540 let mut links = vec![loc.clone()];
541 links.extend(rs);
542 planned.push(Planned {
543 name: name.clone(),
544 schema: schema.clone(),
545 links: links.clone(),
546 });
547 if !is_archive && !name.ends_with("Deleted") {
549 let aobj = archive_defn(obj, name, &schema.collection)
550 .as_object()
551 .cloned()
552 .unwrap();
553 let aname = format!("{}Deleted", name);
554 let acoll = format!("{}_deleted", schema.collection);
555 let aschema = build_schema(&aobj, aname.clone(), acoll, &links[0])?;
556 planned.push(Planned {
557 name: aname,
558 schema: aschema,
559 links,
560 });
561 }
562 } else {
563 let Some(existing) = self.catalog.entries.get(name) else {
565 return Err(format!(
566 "未找到主 schema: {}(从定义 `replica` 必须伴随同批主定义,或该 name 已注册)",
567 name
568 ));
569 };
570 let mut links = existing.links.clone();
571 links.extend(rs);
572 planned.push(Planned {
573 name: name.clone(),
574 schema: existing.schema.clone(),
575 links,
576 });
577 }
578 }
579
580 let batch_names: std::collections::HashSet<&str> =
582 planned.iter().map(|p| p.name.as_str()).collect();
583 let mut quads: Vec<(String, String, Location)> = Vec::new();
584 for (n, e) in &self.catalog.entries {
585 if batch_names.contains(n.as_str()) {
586 continue; }
588 for l in &e.links {
589 quads.push((n.clone(), e.schema.collection.clone(), l.clone()));
590 }
591 }
592 for p in &planned {
593 for l in &p.links {
594 quads.push((p.name.clone(), p.schema.collection.clone(), l.clone()));
595 }
596 }
597 if let Some(c) = detect_location_conflict(&quads) {
598 return Err(format!(
599 "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
600 c.source, c.database, c.schema, c.collection, c.prev
601 ));
602 }
603
604 let catalog = Arc::make_mut(&mut self.catalog); for p in planned {
607 if let Some(e) = catalog.entries.get_mut(&p.name) {
608 e.schema = p.schema;
609 e.links = p.links;
610 e.version += 1;
611 } else {
612 catalog.entries.insert(
613 p.name.clone(),
614 Entry {
615 schema: p.schema,
616 links: p.links,
617 version: 1,
618 },
619 );
620 catalog.order.push(p.name);
621 }
622 }
623
624 Ok(())
625 }
626
627 pub fn apply_link_decls(&mut self, cfg: &LoadConfig) -> Result<(), String> {
636 if cfg.links.is_empty() {
637 return Ok(());
638 }
639 self.ensure_base_catalog()?; let declared: std::collections::HashSet<&str> =
643 cfg.links.iter().map(|(n, _)| n.as_str()).collect();
644 let mut planned: Vec<(String, Vec<Location>)> = Vec::new();
645 for (name, decls) in &cfg.links {
646 let entry = self.get_entry(name).map_err(|_| {
647 format!("未注册 schema: {name}(config.links 引用了不存在的 schema)")
648 })?;
649 let mut links = entry.links.clone();
650 for decl in decls {
651 let target = Location {
652 source: decl.source.clone(),
653 database: decl.database.clone(),
654 schema: decl.schema.clone(),
655 role: decl.role,
656 priority: decl.priority,
657 };
658 match links.iter_mut().find(|l| l.same_place(&target)) {
659 Some(hit) => {
660 hit.role = decl.role;
661 hit.priority = decl.priority;
662 }
663 None => links.push(target),
664 }
665 }
666 if !name.ends_with("Deleted") {
668 let aname = format!("{}Deleted", name);
669 if !declared.contains(aname.as_str()) && self.catalog.entries.contains_key(&aname) {
670 planned.push((aname, links.clone()));
671 }
672 }
673 planned.push((name.clone(), links));
674 }
675
676 let catalog = Arc::make_mut(&mut self.catalog);
678 for (name, links) in planned {
679 if let Some(e) = catalog.entries.get_mut(&name) {
680 e.links = links;
681 }
682 }
683 Ok(())
684 }
685
686 pub fn links_of(&self, name: &str) -> Result<&[Location], String> {
688 Ok(&self.get_entry(name)?.links)
689 }
690
691 pub fn resolve_read_links(
706 &self,
707 name: &str,
708 unreachable: Option<&[String]>,
709 ) -> Result<ReadLinks, String> {
710 let mut links = self.links_of(name)?.to_vec();
711 links.sort_by_key(|l| {
712 (
713 l.role.map_or(4u8, |r| r.rank()),
714 l.priority.unwrap_or(i64::MAX),
715 )
716 });
717
718 let mut degraded: Vec<Value> = Vec::new();
719 if let Some(unreachable) = unreachable {
720 links.retain(|l| {
721 if unreachable.iter().any(|s| s == &l.source) {
722 degraded.push(json!({
723 "code": "readLinkUnreachable",
724 "layer": "schema",
725 "message": format!("schema {} 的链路 {} 不可达,已跳过", name, l.source),
726 "hint": "检查该数据源连接;或调整 store.config.json 的 links 优先级",
727 "schema": name,
728 "source": l.source,
729 }));
730 false
731 } else {
732 true
733 }
734 });
735 }
736
737 if links.is_empty() {
738 return Err(format!(
739 "ERR_READ_NO_REACHABLE_LINK: {} 的全部链路不可达(已按 unreachable 跳过)",
740 name
741 ));
742 }
743 Ok(ReadLinks { links, degraded })
744 }
745
746 pub fn primary_location(&self, name: &str) -> Result<&Location, String> {
748 self.get_entry(name)?
749 .links
750 .first()
751 .ok_or_else(|| format!("Schema 无落点链路: {}", name))
752 }
753
754 pub fn version_of(&self, name: &str) -> Result<u64, String> {
756 Ok(self.get_entry(name)?.version)
757 }
758
759 fn get_entry(&self, name: &str) -> Result<&Entry, String> {
760 self.catalog
761 .entries
762 .get(name)
763 .ok_or_else(|| format!("Schema 未注册: {}", name))
764 }
765
766 pub fn get(&self, name: &str) -> Result<&Schema, String> {
768 self.get_entry(name).map(|e| &e.schema)
769 }
770
771 pub fn has(&self, name: &str) -> bool {
772 self.catalog.entries.contains_key(name)
773 }
774
775 pub fn clear(&mut self) {
784 if let Err(e) = self.ensure_base_catalog() {
785 panic!("{e}");
786 }
787 let catalog = Arc::make_mut(&mut self.catalog);
788 catalog.entries.clear();
789 catalog.order.clear();
790 }
791
792 pub fn set_require_context(&mut self, require: bool) {
796 self.policy.require_context = require;
797 }
798
799 pub fn require_context(&self) -> bool {
801 self.policy.require_context
802 }
803
804 pub fn set_profile(&mut self, profile: Profile) {
809 self.policy.profile = profile;
810 }
811
812 pub fn profile(&self) -> Profile {
814 self.policy.profile
815 }
816
817 pub fn set_exempt_roles(&mut self, roles: Vec<String>) {
819 self.policy.role_rules.exempt_roles = roles;
820 }
821
822 pub fn set_deny_write_roles(&mut self, roles: Vec<String>) {
824 self.policy.role_rules.deny_write_roles = roles;
825 }
826
827 pub fn set_unconfigured_policy(&mut self, policy: crate::permission::UnconfiguredPolicy) {
829 self.policy.role_rules.unconfigured = policy;
830 }
831
832 pub fn role_rules(&self) -> &crate::permission::RoleRules {
834 &self.policy.role_rules
835 }
836
837 pub fn set_meta_policy(&mut self, closed: bool, roles: Vec<String>) {
840 self.policy.meta_policy = crate::permission::MetaPolicy { closed, roles };
841 }
842
843 pub fn meta_policy(&self) -> &crate::permission::MetaPolicy {
845 &self.policy.meta_policy
846 }
847
848 pub fn set_write_link_policy(&mut self, policy: WriteLinkPolicy) {
853 self.policy.write_link_policy = policy;
854 }
855
856 pub fn write_link_policy(&self) -> WriteLinkPolicy {
858 self.policy.write_link_policy
859 }
860
861 pub fn can_register(&self, ctx: Option<&crate::permission::Context>) -> bool {
867 crate::permission::can_register(&self.policy.meta_policy, ctx)
868 }
869
870 pub fn set_rbac(&mut self, policy: Option<&Value>) -> Result<(), String> {
873 match policy {
874 None => {
875 self.policy.rbac = None;
876 Ok(())
877 }
878 Some(v) => {
879 let p = crate::rbac::RbacPolicy::from_json(v)?;
880 self.policy.rbac = Some(p);
881 Ok(())
882 }
883 }
884 }
885
886 pub fn rbac(&self) -> Option<&crate::rbac::RbacPolicy> {
888 self.policy.rbac.as_ref()
889 }
890
891 pub fn get_by_location(
896 &self,
897 source: &str,
898 database: Option<&str>,
899 schema: Option<&str>,
900 collection: &str,
901 ) -> Result<&Schema, String> {
902 let db = database.filter(|s| !s.is_empty());
903 let sc = schema.filter(|s| !s.is_empty());
904 self.catalog
905 .entries
906 .values()
907 .find(|e| {
908 e.schema.collection == collection
909 && e.links.iter().any(|l| {
910 l.source == source
911 && l.database.as_deref().filter(|s| !s.is_empty()) == db
912 && l.schema.as_deref().filter(|s| !s.is_empty()) == sc
913 })
914 })
915 .map(|e| &e.schema)
916 .ok_or_else(|| {
917 format!(
918 "Schema 未定位(source = {}, database = {:?}, schema = {:?}, collection = {})",
919 source, db, sc, collection
920 )
921 })
922 }
923
924 pub fn get_for_command(
931 &self,
932 source: &str,
933 database: Option<&str>,
934 schema: Option<&str>,
935 collection: &str,
936 ) -> Result<&Schema, String> {
937 if let Ok(s) = self.get_by_location(source, database, schema, collection) {
938 return Ok(s);
939 }
940 let candidates: Vec<&Entry> = self
941 .catalog
942 .entries
943 .values()
944 .filter(|e| {
945 e.schema.collection == collection && e.links.iter().any(|l| l.source == source)
946 })
947 .collect();
948 match candidates.len() {
949 0 => self.get_by_location(source, database, schema, collection),
950 1 => Ok(&candidates[0].schema),
951 _ => candidates
952 .iter()
953 .find(|e| e.schema.database().is_none() && e.schema.schema().is_none())
954 .map(|e| &e.schema)
955 .ok_or_else(|| {
956 format!(
957 "命令定位 ({}, {:?}, {:?}, {}) 存在多个结构 schema(多租户 override 需唯一的 (source, collection) 结构声明)",
958 source, database, schema, collection
959 )
960 }),
961 }
962 }
963
964 pub fn list(&self) -> Vec<String> {
965 self.catalog.order.clone()
966 }
967
968 pub fn schema_datasource(&self, name: &str) -> Result<Option<String>, String> {
970 Ok(self.get(name)?.source.clone())
971 }
972
973 pub fn resolve_datasource(
978 &self,
979 schema_name: &str,
980 config: &Value,
981 ) -> Result<crate::datasource::DataSource, String> {
982 let declared = self.get(schema_name)?.source.as_deref();
983 let cfg = crate::datasource::DataSourceConfig::from_json(config)?;
984 cfg.resolve(declared)
985 }
986}
987
988fn build_schema(
991 obj: &Map<String, Value>,
992 name: String,
993 collection: String,
994 loc: &Location,
995) -> Result<Schema, String> {
996 let timestamps = parse_timestamps(obj)?;
997 let mut fields = normalize_fields(obj.get("fields"))?;
998 if timestamps {
999 add_timestamp_fields(&mut fields);
1000 }
1001 let relations = parse_relations(obj);
1002 let computes = parse_computes(obj, &relations)?;
1004 let triggers = super::triggers::parse_triggers(obj, &name, &fields)?;
1006 let mut ids: Vec<&str> = Vec::new();
1009 ids.extend(fields.keys().map(String::as_str));
1010 ids.extend(relations.keys().map(String::as_str));
1011 ids.extend(computes.iter().map(|(k, _)| k.as_str()));
1012 crate::naming::detect_conflicts(ids)?;
1013 Ok(Schema {
1014 name,
1015 collection,
1016 id_prefix: obj
1017 .get("idPrefix")
1018 .and_then(|v| v.as_str())
1019 .unwrap_or("")
1020 .to_string(),
1021 timestamps,
1022 fields,
1023 relations,
1024 computes,
1025 triggers,
1026 read: str_list(obj.get("read")),
1027 write: str_list(obj.get("write")),
1028 indexes: obj
1029 .get("indexes")
1030 .and_then(|v| v.as_array())
1031 .cloned()
1032 .unwrap_or_default(),
1033 source: if loc.source == crate::datasource::DEFAULT_SOURCE {
1034 None
1035 } else {
1036 Some(loc.source.clone())
1037 },
1038 database: loc.database.clone().filter(|s| !s.is_empty()),
1039 schema: loc.schema.clone().filter(|s| !s.is_empty()),
1040 })
1041}
1042
1043fn parse_timestamps(obj: &Map<String, Value>) -> Result<bool, String> {
1045 match obj.get("timestamps") {
1046 None | Some(Value::Null) | Some(Value::Bool(true)) => Ok(true),
1047 Some(Value::Bool(false)) => Ok(false),
1048 Some(Value::String(s)) if s == "ms" || s == "s" => Ok(true),
1049 Some(Value::String(s)) => Err(format!(
1050 "timestamps 仅支持 true/false/'ms'/'s',实际 {:?}",
1051 s
1052 )),
1053 Some(other) => Err(format!(
1054 "timestamps 仅支持 true/false/'ms'/'s',实际 {}",
1055 other
1056 )),
1057 }
1058}
1059
1060fn add_timestamp_fields(fields: &mut HashMap<String, FieldDef>) {
1062 for key in ["createdAt", "updatedAt"] {
1063 fields.entry(key.to_string()).or_insert_with(|| FieldDef {
1064 field_type: "number".to_string(),
1065 required: false,
1066 default: None,
1067 read: None,
1068 write: None,
1069 fields: None,
1070 strategy: None,
1071 });
1072 }
1073}
1074
1075fn parse_relations(obj: &Map<String, Value>) -> HashMap<String, RelationDef> {
1076 let mut relations = HashMap::new();
1077 if let Some(Value::Object(rm)) = obj.get("relations") {
1078 for (key, val) in rm {
1079 let vo = val.as_object();
1080 let get_str = |k: &str| -> Option<String> {
1081 vo.and_then(|o| o.get(k))
1082 .and_then(|v| v.as_str())
1083 .filter(|s| !s.is_empty())
1084 .map(String::from)
1085 };
1086 relations.insert(
1087 key.clone(),
1088 RelationDef {
1089 model: get_str("model").unwrap_or_default(),
1090 rel_type: get_str("type").unwrap_or_else(|| "many".to_string()),
1091 local_field: get_str("localField").unwrap_or_else(|| "_id".to_string()),
1092 foreign_field: get_str("foreignField").unwrap_or_else(|| key.clone()),
1093 read: str_list(vo.and_then(|o| o.get("read"))),
1094 },
1095 );
1096 }
1097 }
1098 relations
1099}
1100
1101fn parse_computes(
1102 obj: &Map<String, Value>,
1103 relations: &HashMap<String, RelationDef>,
1104) -> Result<Vec<(String, ComputeDef)>, String> {
1105 let mut computes = Vec::new();
1106 if let Some(Value::Object(cm)) = obj.get("computes") {
1107 for (key, val) in cm {
1108 let vo = val.as_object();
1109 let get = |k: &str| vo.and_then(|o| o.get(k)).cloned();
1110 let has_fn = get("fn").map(|v| is_truthy(&v)).unwrap_or(false);
1111 let has_async_fn = get("asyncFn").map(|v| is_truthy(&v)).unwrap_or(false);
1112 let agg = parse_agg(key, get("agg").as_ref(), relations, has_fn || has_async_fn)?;
1113 computes.push((
1114 key.clone(),
1115 ComputeDef {
1116 comp_type: get("type")
1117 .and_then(|v| v.as_str().map(String::from))
1118 .filter(|s| !s.is_empty())
1119 .unwrap_or_else(|| "any".to_string()),
1120 has_fn,
1121 has_async_fn,
1122 fn_ref: get("fnRef")
1123 .and_then(|v| v.as_str().map(String::from))
1124 .filter(|s| !s.is_empty()),
1125 agg,
1126 depends: str_list(get("depends").as_ref()).unwrap_or_default(),
1127 read: str_list(get("read").as_ref()),
1128 },
1129 ));
1130 }
1131 }
1132 Ok(computes)
1133}
1134
1135fn parse_agg(
1143 key: &str,
1144 raw: Option<&Value>,
1145 relations: &HashMap<String, RelationDef>,
1146 has_callback: bool,
1147) -> Result<Option<Value>, String> {
1148 let Some(v) = raw.filter(|v| is_truthy(v)) else {
1149 return Ok(None);
1150 };
1151 if has_callback {
1152 return Err(format!(
1153 "计算列 \"{key}\" 不能同时声明 agg 与 fn/asyncFn(二选一)"
1154 ));
1155 }
1156 let obj = v.as_object().ok_or_else(|| {
1157 format!("计算列 \"{key}\" 的 agg 必须是对象,如 {{\"$count\": \"lessons\"}}")
1158 })?;
1159 let mut it = obj.iter();
1160 let (op, arg) = match (it.next(), it.next()) {
1161 (Some(kv), None) => kv,
1162 _ => return Err(format!("计算列 \"{key}\" 的 agg 必须且只能声明一个算子键")),
1163 };
1164 if !AGG_OPS.contains(&op.as_str()) {
1165 return Err(format!(
1166 "计算列 \"{key}\" 的 agg 算子 {op} 不在白名单($count/$sum/$avg/$min/$max)"
1167 ));
1168 }
1169 let path = arg
1170 .as_str()
1171 .filter(|s| !s.is_empty() && *s != "*")
1172 .ok_or_else(|| {
1173 format!("计算列 \"{key}\" 的 agg 取值必须是关系路径字符串(如 \"lessons\" / \"lessons.duration\")")
1174 })?;
1175 let (rel, field) = match path.split_once('.') {
1176 Some((r, f)) => (r, Some(f)),
1177 None => (path, None),
1178 };
1179 if !relations.contains_key(rel) {
1180 return Err(format!(
1181 "计算列 \"{key}\" 的 agg 引用的关系 \"{rel}\" 未在 schema 中定义"
1182 ));
1183 }
1184 if field
1185 .map(|f| f.is_empty() || f.contains('.'))
1186 .unwrap_or(false)
1187 {
1188 return Err(format!(
1189 "计算列 \"{key}\" 的 agg 关系路径 \"{path}\" 非法:仅支持单级「关系.字段」(二级路径首批不支持)"
1190 ));
1191 }
1192 match op.as_str() {
1193 "$count" if field.is_some() => Err(format!(
1194 "计算列 \"{key}\" 的 $count 仅支持关系整行计数(如 \"lessons\"),按字段计数待 P5"
1195 )),
1196 "$count" => Ok(Some(v.clone())),
1197 _ if field.is_none() => Err(format!(
1198 "计算列 \"{key}\" 的 agg 算子 {op} 必须引用关系字段(如 \"lessons.duration\")"
1199 )),
1200 _ => Ok(Some(v.clone())),
1201 }
1202}
1203
1204fn archive_defn(obj: &Map<String, Value>, name: &str, collection: &str) -> Value {
1207 let mut arch_fields = obj
1208 .get("fields")
1209 .and_then(|v| v.as_object())
1210 .cloned()
1211 .unwrap_or_default();
1212 arch_fields.insert("deletedAt".to_string(), json!({ "type": "number" }));
1213 if let Some(idf) = arch_fields.get_mut("_id") {
1216 if let Some(o) = idf.as_object_mut() {
1217 o.remove("strategy");
1218 }
1219 }
1220 let arch = json!({
1221 "name": format!("{}Deleted", name),
1222 "collection": format!("{}_deleted", collection),
1223 "idPrefix": "",
1224 "_isArchive": true,
1225 "fields": Value::Object(arch_fields),
1226 "indexes": obj.get("indexes").cloned().unwrap_or_else(|| json!([])),
1227 });
1228 arch
1229}
1230
1231#[cfg(test)]
1232mod tests {
1233 use super::*;
1234
1235 #[test]
1236 fn write_link_policy_defaults_to_reject() {
1237 let r = Registry::new();
1238 assert_eq!(r.write_link_policy(), WriteLinkPolicy::Reject);
1239 let mut r2 = Registry::new();
1240 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::Reject);
1241 r2.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
1242 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
1243 }
1244
1245 use crate::permission::UnconfiguredPolicy;
1248
1249 fn overrides_all() -> PolicyOverrides {
1251 PolicyOverrides::from_value(&json!({
1252 "profile": "standard",
1253 "requireContext": false,
1254 "rbac": null,
1255 "roleRules": { "exemptRoles": ["x"], "denyWriteRoles": ["y"], "unconfigured": "open" },
1256 "metaPolicy": { "closed": false, "roles": [] },
1257 "writeLinkPolicy": "reject"
1258 }))
1259 .expect("合法覆盖 JSON")
1260 }
1261
1262 #[test]
1264 fn with_policy_override_matrix() {
1265 let mut base = Registry::new();
1266 base.set_require_context(true);
1267 base.set_profile(Profile::Text2Query);
1268 base.set_rbac(Some(&json!({}))).expect("合法 RBAC");
1269 base.set_exempt_roles(vec!["a".to_string()]);
1270 base.set_deny_write_roles(vec!["b".to_string()]);
1271 base.set_unconfigured_policy(UnconfiguredPolicy::Closed);
1272 base.set_meta_policy(true, vec!["m".to_string()]);
1273 base.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
1274
1275 let inherit = base.with_policy(&PolicyOverrides::default());
1277 assert!(inherit.is_view(), "派生视图 is_view 应为真");
1278 assert!(inherit.require_context());
1279 assert_eq!(inherit.profile(), Profile::Text2Query);
1280 assert!(inherit.rbac().is_some());
1281 assert_eq!(inherit.role_rules().exempt_roles, vec!["a".to_string()]);
1282 assert_eq!(inherit.role_rules().deny_write_roles, vec!["b".to_string()]);
1283 assert_eq!(
1284 inherit.role_rules().unconfigured,
1285 UnconfiguredPolicy::Closed
1286 );
1287 assert!(inherit.meta_policy().closed);
1288 assert_eq!(inherit.meta_policy().roles, vec!["m".to_string()]);
1289 assert_eq!(inherit.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
1290
1291 let full = base.with_policy(&overrides_all());
1293 assert!(!full.require_context());
1294 assert_eq!(full.profile(), Profile::Standard);
1295 assert!(full.rbac().is_none(), "rbac:null 应清除");
1296 assert_eq!(full.role_rules().exempt_roles, vec!["x".to_string()]);
1297 assert_eq!(full.role_rules().deny_write_roles, vec!["y".to_string()]);
1298 assert_eq!(full.role_rules().unconfigured, UnconfiguredPolicy::Open);
1299 assert!(!full.meta_policy().closed);
1300 assert_eq!(full.write_link_policy(), WriteLinkPolicy::Reject);
1301
1302 assert!(base.require_context());
1304 assert_eq!(base.profile(), Profile::Text2Query);
1305 assert!(base.rbac().is_some());
1306 assert!(base.meta_policy().closed);
1307 }
1308
1309 #[test]
1311 fn view_register_is_readonly_err() {
1312 let base = Registry::new();
1313 let mut view = base.with_policy(&PolicyOverrides::default());
1314 let err = view
1315 .register(&json!({ "name": "A", "fields": {} }))
1316 .expect_err("视图注册应 Err");
1317 assert!(
1318 err.starts_with(ERR_POLICY_VIEW_READONLY),
1319 "前缀可匹配: {err}"
1320 );
1321 let err2 = view
1322 .register_with_ctx(&json!({ "name": "A", "fields": {} }), None)
1323 .expect_err("视图注册应 Err");
1324 assert!(
1325 err2.starts_with(ERR_POLICY_VIEW_READONLY),
1326 "前缀可匹配: {err2}"
1327 );
1328 let err3 = view
1329 .register_batch(
1330 &[(json!({ "name": "A", "fields": {} }), Location::default())],
1331 None,
1332 )
1333 .expect_err("视图批量注册应 Err");
1334 assert!(
1335 err3.starts_with(ERR_POLICY_VIEW_READONLY),
1336 "前缀可匹配: {err3}"
1337 );
1338 }
1339
1340 #[test]
1342 #[should_panic(expected = "ERR_POLICY_VIEW_READONLY")]
1343 fn view_clear_panics_readonly() {
1344 let base = Registry::new();
1345 let mut view = base.with_policy(&PolicyOverrides::default());
1346 view.clear();
1347 }
1348
1349 #[test]
1351 fn view_sees_catalog_snapshot() {
1352 let mut base = Registry::new();
1353 base.register(&json!({ "name": "A", "fields": { "x": { "type": "number" } } }))
1354 .expect("base 注册成功");
1355 let view = base.with_policy(&PolicyOverrides::default());
1356 assert!(view.has("A"), "派生视图应见 base 已注册目录");
1357 assert_eq!(view.get("A").expect("视图可取").name, "A");
1358 assert_eq!(view.list(), base.list());
1359
1360 let mut base2 = Registry::new();
1362 base2
1363 .register(&json!({ "name": "A", "fields": {} }))
1364 .unwrap();
1365 let snap = base2.with_policy(&PolicyOverrides::default());
1366 base2
1367 .register(&json!({ "name": "B", "fields": {} }))
1368 .unwrap();
1369 assert!(snap.has("A"), "在途视图仍持其快照");
1370 let fresh = base2.with_policy(&PolicyOverrides::default());
1371 assert!(fresh.has("B"), "新派生视图应见 base 后续注册");
1372 }
1373
1374 #[test]
1376 fn view_policy_set_isolated() {
1377 let base = Registry::new();
1378 let mut view = base.with_policy(&PolicyOverrides::default());
1379 view.set_profile(Profile::Text2Query);
1380 assert_eq!(view.profile(), Profile::Text2Query);
1381 assert_eq!(base.profile(), Profile::Standard, "视图 set 不影响 base");
1382 let sibling = base.with_policy(&PolicyOverrides::default());
1383 assert_eq!(sibling.profile(), Profile::Standard, "不影响兄弟视图");
1384 }
1385
1386 #[test]
1388 fn from_value_unknown_key_and_rbac_null() {
1389 let e = PolicyOverrides::from_value(&json!({ "nope": 1 })).expect_err("未知键应 Err");
1390 assert!(e.to_string().starts_with(ERR_POLICY_VIEW_READONLY), "{e}");
1391 assert!(
1392 PolicyOverrides::from_value(&json!({ "profile": "bogus" })).is_err(),
1393 "非法 profile 应 Err(禁静默回落)"
1394 );
1395 assert!(
1396 PolicyOverrides::from_value(&json!({ "writeLinkPolicy": "bogus" })).is_err(),
1397 "非法 writeLinkPolicy 应 Err"
1398 );
1399 assert!(
1400 PolicyOverrides::from_value(&json!({})).is_ok(),
1401 "空对象合法"
1402 );
1403
1404 let mut base = Registry::new();
1405 base.set_rbac(Some(&json!({}))).unwrap();
1406 let cleared =
1407 base.with_policy(&PolicyOverrides::from_value(&json!({ "rbac": null })).unwrap());
1408 assert!(cleared.rbac().is_none(), "rbac:null 应清除");
1409 let kept = base.with_policy(&PolicyOverrides::from_value(&json!({})).unwrap());
1410 assert!(kept.rbac().is_some(), "缺键应继承");
1411 assert!(base.rbac().is_some(), "base 不受影响");
1412 }
1413}