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::{
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 if !is_archive && !name.ends_with("Deleted") {
551 let aobj = archive_defn(obj, name, &schema.collection)
552 .as_object()
553 .cloned()
554 .unwrap();
555 let aname = format!("{}Deleted", name);
556 let acoll = format!("{}_deleted", schema.collection);
557 let aschema = build_schema(&aobj, aname.clone(), acoll, &links[0])?;
558 planned.push(Planned {
559 name: aname,
560 schema: aschema,
561 links,
562 });
563 }
564 } else {
565 let Some(existing) = self.catalog.entries.get(name) else {
567 return Err(format!(
568 "未找到主 schema: {}(从定义 `replica` 必须伴随同批主定义,或该 name 已注册)",
569 name
570 ));
571 };
572 let mut links = existing.links.clone();
573 links.extend(rs);
574 planned.push(Planned {
575 name: name.clone(),
576 schema: existing.schema.clone(),
577 links,
578 });
579 }
580 }
581
582 let batch_names: std::collections::HashSet<&str> =
584 planned.iter().map(|p| p.name.as_str()).collect();
585 let mut quads: Vec<(String, String, Location)> = Vec::new();
586 for (n, e) in &self.catalog.entries {
587 if batch_names.contains(n.as_str()) {
588 continue; }
590 for l in &e.links {
591 quads.push((n.clone(), e.schema.collection.clone(), l.clone()));
592 }
593 }
594 for p in &planned {
595 for l in &p.links {
596 quads.push((p.name.clone(), p.schema.collection.clone(), l.clone()));
597 }
598 }
599 if let Some(c) = detect_location_conflict(&quads) {
600 return Err(format!(
601 "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
602 c.source, c.database, c.schema, c.collection, c.prev
603 ));
604 }
605
606 let catalog = Arc::make_mut(&mut self.catalog); for p in planned {
609 if let Some(e) = catalog.entries.get_mut(&p.name) {
610 e.schema = p.schema;
611 e.links = p.links;
612 e.version += 1;
613 } else {
614 catalog.entries.insert(
615 p.name.clone(),
616 Entry {
617 schema: p.schema,
618 links: p.links,
619 version: 1,
620 },
621 );
622 catalog.order.push(p.name);
623 }
624 }
625
626 Ok(())
627 }
628
629 pub fn apply_link_decls(&mut self, cfg: &LoadConfig) -> Result<(), String> {
638 if cfg.links.is_empty() {
639 return Ok(());
640 }
641 self.ensure_base_catalog()?; let declared: std::collections::HashSet<&str> =
645 cfg.links.iter().map(|(n, _)| n.as_str()).collect();
646 let mut planned: Vec<(String, Vec<Location>)> = Vec::new();
647 for (name, decls) in &cfg.links {
648 let entry = self.get_entry(name).map_err(|_| {
649 format!("未注册 schema: {name}(config.links 引用了不存在的 schema)")
650 })?;
651 let mut links = entry.links.clone();
652 for decl in decls {
653 let target = Location {
654 source: decl.source.clone(),
655 database: decl.database.clone(),
656 schema: decl.schema.clone(),
657 role: decl.role,
658 priority: decl.priority,
659 };
660 match links.iter_mut().find(|l| l.same_place(&target)) {
661 Some(hit) => {
662 hit.role = decl.role;
663 hit.priority = decl.priority;
664 }
665 None => links.push(target),
666 }
667 }
668 if !name.ends_with("Deleted") {
670 let aname = format!("{}Deleted", name);
671 if !declared.contains(aname.as_str()) && self.catalog.entries.contains_key(&aname) {
672 planned.push((aname, links.clone()));
673 }
674 }
675 planned.push((name.clone(), links));
676 }
677
678 let catalog = Arc::make_mut(&mut self.catalog);
680 for (name, links) in planned {
681 if let Some(e) = catalog.entries.get_mut(&name) {
682 e.links = links;
683 }
684 }
685 Ok(())
686 }
687
688 pub fn links_of(&self, name: &str) -> Result<&[Location], String> {
690 Ok(&self.get_entry(name)?.links)
691 }
692
693 pub fn resolve_read_links(
708 &self,
709 name: &str,
710 unreachable: Option<&[String]>,
711 ) -> Result<ReadLinks, String> {
712 let mut links = self.links_of(name)?.to_vec();
713 links.sort_by_key(|l| {
714 (
715 l.role.map_or(4u8, |r| r.rank()),
716 l.priority.unwrap_or(i64::MAX),
717 )
718 });
719
720 let mut degraded: Vec<Value> = Vec::new();
721 if let Some(unreachable) = unreachable {
722 links.retain(|l| {
723 if unreachable.iter().any(|s| s == &l.source) {
724 degraded.push(json!({
725 "code": "readLinkUnreachable",
726 "layer": "schema",
727 "message": format!("schema {} 的链路 {} 不可达,已跳过", name, l.source),
728 "hint": "检查该数据源连接;或调整 store.config.json 的 links 优先级",
729 "schema": name,
730 "source": l.source,
731 }));
732 false
733 } else {
734 true
735 }
736 });
737 }
738
739 if links.is_empty() {
740 return Err(format!(
741 "ERR_READ_NO_REACHABLE_LINK: {} 的全部链路不可达(已按 unreachable 跳过)",
742 name
743 ));
744 }
745 Ok(ReadLinks { links, degraded })
746 }
747
748 pub fn primary_location(&self, name: &str) -> Result<&Location, String> {
750 self.get_entry(name)?
751 .links
752 .first()
753 .ok_or_else(|| format!("Schema 无落点链路: {}", name))
754 }
755
756 pub fn version_of(&self, name: &str) -> Result<u64, String> {
758 Ok(self.get_entry(name)?.version)
759 }
760
761 fn get_entry(&self, name: &str) -> Result<&Entry, String> {
762 self.catalog
763 .entries
764 .get(name)
765 .ok_or_else(|| format!("Schema 未注册: {}", name))
766 }
767
768 pub fn get(&self, name: &str) -> Result<&Schema, String> {
770 self.get_entry(name).map(|e| &e.schema)
771 }
772
773 pub fn has(&self, name: &str) -> bool {
774 self.catalog.entries.contains_key(name)
775 }
776
777 pub fn clear(&mut self) {
786 if let Err(e) = self.ensure_base_catalog() {
787 panic!("{e}");
788 }
789 let catalog = Arc::make_mut(&mut self.catalog);
790 catalog.entries.clear();
791 catalog.order.clear();
792 }
793
794 pub fn set_require_context(&mut self, require: bool) {
798 self.policy.require_context = require;
799 }
800
801 pub fn require_context(&self) -> bool {
803 self.policy.require_context
804 }
805
806 pub fn set_profile(&mut self, profile: Profile) {
811 self.policy.profile = profile;
812 }
813
814 pub fn profile(&self) -> Profile {
816 self.policy.profile
817 }
818
819 pub fn set_exempt_roles(&mut self, roles: Vec<String>) {
821 self.policy.role_rules.exempt_roles = roles;
822 }
823
824 pub fn set_deny_write_roles(&mut self, roles: Vec<String>) {
826 self.policy.role_rules.deny_write_roles = roles;
827 }
828
829 pub fn set_unconfigured_policy(&mut self, policy: crate::permission::UnconfiguredPolicy) {
831 self.policy.role_rules.unconfigured = policy;
832 }
833
834 pub fn role_rules(&self) -> &crate::permission::RoleRules {
836 &self.policy.role_rules
837 }
838
839 pub fn set_meta_policy(&mut self, closed: bool, roles: Vec<String>) {
842 self.policy.meta_policy = crate::permission::MetaPolicy { closed, roles };
843 }
844
845 pub fn meta_policy(&self) -> &crate::permission::MetaPolicy {
847 &self.policy.meta_policy
848 }
849
850 pub fn set_write_link_policy(&mut self, policy: WriteLinkPolicy) {
855 self.policy.write_link_policy = policy;
856 }
857
858 pub fn write_link_policy(&self) -> WriteLinkPolicy {
860 self.policy.write_link_policy
861 }
862
863 pub fn can_register(&self, ctx: Option<&crate::permission::Context>) -> bool {
869 crate::permission::can_register(&self.policy.meta_policy, ctx)
870 }
871
872 pub fn set_rbac(&mut self, policy: Option<&Value>) -> Result<(), String> {
875 match policy {
876 None => {
877 self.policy.rbac = None;
878 Ok(())
879 }
880 Some(v) => {
881 let p = crate::rbac::RbacPolicy::from_json(v)?;
882 self.policy.rbac = Some(p);
883 Ok(())
884 }
885 }
886 }
887
888 pub fn rbac(&self) -> Option<&crate::rbac::RbacPolicy> {
890 self.policy.rbac.as_ref()
891 }
892
893 pub fn get_by_location(
898 &self,
899 source: &str,
900 database: Option<&str>,
901 schema: Option<&str>,
902 collection: &str,
903 ) -> Result<&Schema, String> {
904 let db = database.filter(|s| !s.is_empty());
905 let sc = schema.filter(|s| !s.is_empty());
906 self.catalog
907 .entries
908 .values()
909 .find(|e| {
910 e.schema.collection == collection
911 && e.links.iter().any(|l| {
912 l.source == source
913 && l.database.as_deref().filter(|s| !s.is_empty()) == db
914 && l.schema.as_deref().filter(|s| !s.is_empty()) == sc
915 })
916 })
917 .map(|e| &e.schema)
918 .ok_or_else(|| {
919 format!(
920 "Schema 未定位(source = {}, database = {:?}, schema = {:?}, collection = {})",
921 source, db, sc, collection
922 )
923 })
924 }
925
926 pub fn get_for_command(
933 &self,
934 source: &str,
935 database: Option<&str>,
936 schema: Option<&str>,
937 collection: &str,
938 ) -> Result<&Schema, String> {
939 if let Ok(s) = self.get_by_location(source, database, schema, collection) {
940 return Ok(s);
941 }
942 let candidates: Vec<&Entry> = self
943 .catalog
944 .entries
945 .values()
946 .filter(|e| {
947 e.schema.collection == collection && e.links.iter().any(|l| l.source == source)
948 })
949 .collect();
950 match candidates.len() {
951 0 => self.get_by_location(source, database, schema, collection),
952 1 => Ok(&candidates[0].schema),
953 _ => candidates
954 .iter()
955 .find(|e| e.schema.database().is_none() && e.schema.schema().is_none())
956 .map(|e| &e.schema)
957 .ok_or_else(|| {
958 format!(
959 "命令定位 ({}, {:?}, {:?}, {}) 存在多个结构 schema(多租户 override 需唯一的 (source, collection) 结构声明)",
960 source, database, schema, collection
961 )
962 }),
963 }
964 }
965
966 pub fn list(&self) -> Vec<String> {
967 self.catalog.order.clone()
968 }
969
970 pub fn schema_datasource(&self, name: &str) -> Result<Option<String>, String> {
972 Ok(self.get(name)?.source.clone())
973 }
974
975 pub fn resolve_datasource(
980 &self,
981 schema_name: &str,
982 config: &Value,
983 ) -> Result<crate::datasource::DataSource, String> {
984 let declared = self.get(schema_name)?.source.as_deref();
985 let cfg = crate::datasource::DataSourceConfig::from_json(config)?;
986 cfg.resolve(declared)
987 }
988}
989
990fn build_schema(
993 obj: &Map<String, Value>,
994 name: String,
995 collection: String,
996 loc: &Location,
997) -> Result<Schema, String> {
998 let timestamps = parse_timestamps(obj)?;
999 let mut fields = normalize_fields(obj.get("fields"))?;
1000 validate_vector_fields(&fields)?;
1002 if timestamps {
1003 add_timestamp_fields(&mut fields);
1004 }
1005 let relations = parse_relations(obj);
1006 let computes = parse_computes(obj, &relations)?;
1008 let triggers = super::triggers::parse_triggers(obj, &name, &fields)?;
1010 for (k, ns) in fields
1015 .keys()
1016 .map(|k| (k, "fields"))
1017 .chain(relations.keys().map(|k| (k, "relations")))
1018 .chain(computes.iter().map(|(k, _)| (k, "computes")))
1019 {
1020 if k == "_score" {
1021 return Err(format!(
1022 "{}名称 \"_score\" 为向量检索评分保留键,不得在 schema {ns} 中声明",
1023 crate::naming::ERR_NAME_CONFLICT
1024 ));
1025 }
1026 }
1027 let mut ids: Vec<&str> = Vec::new();
1030 ids.extend(fields.keys().map(String::as_str));
1031 ids.extend(relations.keys().map(String::as_str));
1032 ids.extend(computes.iter().map(|(k, _)| k.as_str()));
1033 crate::naming::detect_conflicts(ids)?;
1034 Ok(Schema {
1035 name,
1036 collection,
1037 id_prefix: obj
1038 .get("idPrefix")
1039 .and_then(|v| v.as_str())
1040 .unwrap_or("")
1041 .to_string(),
1042 timestamps,
1043 fields,
1044 relations,
1045 computes,
1046 triggers,
1047 read: str_list(obj.get("read")),
1048 write: str_list(obj.get("write")),
1049 indexes: obj
1050 .get("indexes")
1051 .and_then(|v| v.as_array())
1052 .cloned()
1053 .unwrap_or_default(),
1054 source: if loc.source == crate::datasource::DEFAULT_SOURCE {
1055 None
1056 } else {
1057 Some(loc.source.clone())
1058 },
1059 database: loc.database.clone().filter(|s| !s.is_empty()),
1060 schema: loc.schema.clone().filter(|s| !s.is_empty()),
1061 })
1062}
1063
1064fn parse_timestamps(obj: &Map<String, Value>) -> Result<bool, String> {
1066 match obj.get("timestamps") {
1067 None | Some(Value::Null) | Some(Value::Bool(true)) => Ok(true),
1068 Some(Value::Bool(false)) => Ok(false),
1069 Some(Value::String(s)) if s == "ms" || s == "s" => Ok(true),
1070 Some(Value::String(s)) => Err(format!(
1071 "timestamps 仅支持 true/false/'ms'/'s',实际 {:?}",
1072 s
1073 )),
1074 Some(other) => Err(format!(
1075 "timestamps 仅支持 true/false/'ms'/'s',实际 {}",
1076 other
1077 )),
1078 }
1079}
1080
1081fn add_timestamp_fields(fields: &mut HashMap<String, FieldDef>) {
1083 for key in ["createdAt", "updatedAt"] {
1084 fields.entry(key.to_string()).or_insert_with(|| FieldDef {
1085 field_type: "number".to_string(),
1086 required: false,
1087 default: None,
1088 read: None,
1089 write: None,
1090 fields: None,
1091 strategy: None,
1092 dim: None,
1093 metric: None,
1094 });
1095 }
1096}
1097
1098fn parse_relations(obj: &Map<String, Value>) -> HashMap<String, RelationDef> {
1099 let mut relations = HashMap::new();
1100 if let Some(Value::Object(rm)) = obj.get("relations") {
1101 for (key, val) in rm {
1102 let vo = val.as_object();
1103 let get_str = |k: &str| -> Option<String> {
1104 vo.and_then(|o| o.get(k))
1105 .and_then(|v| v.as_str())
1106 .filter(|s| !s.is_empty())
1107 .map(String::from)
1108 };
1109 relations.insert(
1110 key.clone(),
1111 RelationDef {
1112 model: get_str("model").unwrap_or_default(),
1113 rel_type: get_str("type").unwrap_or_else(|| "many".to_string()),
1114 local_field: get_str("localField").unwrap_or_else(|| "_id".to_string()),
1115 foreign_field: get_str("foreignField").unwrap_or_else(|| key.clone()),
1116 read: str_list(vo.and_then(|o| o.get("read"))),
1117 },
1118 );
1119 }
1120 }
1121 relations
1122}
1123
1124fn parse_computes(
1125 obj: &Map<String, Value>,
1126 relations: &HashMap<String, RelationDef>,
1127) -> Result<Vec<(String, ComputeDef)>, String> {
1128 let mut computes = Vec::new();
1129 if let Some(Value::Object(cm)) = obj.get("computes") {
1130 for (key, val) in cm {
1131 let vo = val.as_object();
1132 let get = |k: &str| vo.and_then(|o| o.get(k)).cloned();
1133 let has_fn = get("fn").map(|v| is_truthy(&v)).unwrap_or(false);
1134 let has_async_fn = get("asyncFn").map(|v| is_truthy(&v)).unwrap_or(false);
1135 let agg = parse_agg(key, get("agg").as_ref(), relations, has_fn || has_async_fn)?;
1136 computes.push((
1137 key.clone(),
1138 ComputeDef {
1139 comp_type: get("type")
1140 .and_then(|v| v.as_str().map(String::from))
1141 .filter(|s| !s.is_empty())
1142 .unwrap_or_else(|| "any".to_string()),
1143 has_fn,
1144 has_async_fn,
1145 fn_ref: get("fnRef")
1146 .and_then(|v| v.as_str().map(String::from))
1147 .filter(|s| !s.is_empty()),
1148 agg,
1149 depends: str_list(get("depends").as_ref()).unwrap_or_default(),
1150 read: str_list(get("read").as_ref()),
1151 },
1152 ));
1153 }
1154 }
1155 Ok(computes)
1156}
1157
1158fn parse_agg(
1166 key: &str,
1167 raw: Option<&Value>,
1168 relations: &HashMap<String, RelationDef>,
1169 has_callback: bool,
1170) -> Result<Option<Value>, String> {
1171 let Some(v) = raw.filter(|v| is_truthy(v)) else {
1172 return Ok(None);
1173 };
1174 if has_callback {
1175 return Err(format!(
1176 "计算列 \"{key}\" 不能同时声明 agg 与 fn/asyncFn(二选一)"
1177 ));
1178 }
1179 let obj = v.as_object().ok_or_else(|| {
1180 format!("计算列 \"{key}\" 的 agg 必须是对象,如 {{\"$count\": \"lessons\"}}")
1181 })?;
1182 let mut it = obj.iter();
1183 let (op, arg) = match (it.next(), it.next()) {
1184 (Some(kv), None) => kv,
1185 _ => return Err(format!("计算列 \"{key}\" 的 agg 必须且只能声明一个算子键")),
1186 };
1187 if !AGG_OPS.contains(&op.as_str()) {
1188 return Err(format!(
1189 "计算列 \"{key}\" 的 agg 算子 {op} 不在白名单($count/$sum/$avg/$min/$max)"
1190 ));
1191 }
1192 let path = arg
1193 .as_str()
1194 .filter(|s| !s.is_empty() && *s != "*")
1195 .ok_or_else(|| {
1196 format!("计算列 \"{key}\" 的 agg 取值必须是关系路径字符串(如 \"lessons\" / \"lessons.duration\")")
1197 })?;
1198 let (rel, field) = match path.split_once('.') {
1199 Some((r, f)) => (r, Some(f)),
1200 None => (path, None),
1201 };
1202 if !relations.contains_key(rel) {
1203 return Err(format!(
1204 "计算列 \"{key}\" 的 agg 引用的关系 \"{rel}\" 未在 schema 中定义"
1205 ));
1206 }
1207 if field
1208 .map(|f| f.is_empty() || f.contains('.'))
1209 .unwrap_or(false)
1210 {
1211 return Err(format!(
1212 "计算列 \"{key}\" 的 agg 关系路径 \"{path}\" 非法:仅支持单级「关系.字段」(二级路径首批不支持)"
1213 ));
1214 }
1215 match op.as_str() {
1216 "$count" if field.is_some() => Err(format!(
1217 "计算列 \"{key}\" 的 $count 仅支持关系整行计数(如 \"lessons\"),按字段计数待 P5"
1218 )),
1219 "$count" => Ok(Some(v.clone())),
1220 _ if field.is_none() => Err(format!(
1221 "计算列 \"{key}\" 的 agg 算子 {op} 必须引用关系字段(如 \"lessons.duration\")"
1222 )),
1223 _ => Ok(Some(v.clone())),
1224 }
1225}
1226
1227fn archive_defn(obj: &Map<String, Value>, name: &str, collection: &str) -> Value {
1230 let mut arch_fields = obj
1231 .get("fields")
1232 .and_then(|v| v.as_object())
1233 .cloned()
1234 .unwrap_or_default();
1235 arch_fields.insert("deletedAt".to_string(), json!({ "type": "number" }));
1236 if let Some(idf) = arch_fields.get_mut("_id") {
1239 if let Some(o) = idf.as_object_mut() {
1240 o.remove("strategy");
1241 }
1242 }
1243 let arch = json!({
1244 "name": format!("{}Deleted", name),
1245 "collection": format!("{}_deleted", collection),
1246 "idPrefix": "",
1247 "_isArchive": true,
1248 "fields": Value::Object(arch_fields),
1249 "indexes": obj.get("indexes").cloned().unwrap_or_else(|| json!([])),
1250 });
1251 arch
1252}
1253
1254#[cfg(test)]
1255mod tests {
1256 use super::*;
1257
1258 #[test]
1259 fn write_link_policy_defaults_to_reject() {
1260 let r = Registry::new();
1261 assert_eq!(r.write_link_policy(), WriteLinkPolicy::Reject);
1262 let mut r2 = Registry::new();
1263 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::Reject);
1264 r2.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
1265 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
1266 }
1267
1268 use crate::permission::UnconfiguredPolicy;
1271
1272 fn overrides_all() -> PolicyOverrides {
1274 PolicyOverrides::from_value(&json!({
1275 "profile": "standard",
1276 "requireContext": false,
1277 "rbac": null,
1278 "roleRules": { "exemptRoles": ["x"], "denyWriteRoles": ["y"], "unconfigured": "open" },
1279 "metaPolicy": { "closed": false, "roles": [] },
1280 "writeLinkPolicy": "reject"
1281 }))
1282 .expect("合法覆盖 JSON")
1283 }
1284
1285 #[test]
1287 fn with_policy_override_matrix() {
1288 let mut base = Registry::new();
1289 base.set_require_context(true);
1290 base.set_profile(Profile::Text2Query);
1291 base.set_rbac(Some(&json!({}))).expect("合法 RBAC");
1292 base.set_exempt_roles(vec!["a".to_string()]);
1293 base.set_deny_write_roles(vec!["b".to_string()]);
1294 base.set_unconfigured_policy(UnconfiguredPolicy::Closed);
1295 base.set_meta_policy(true, vec!["m".to_string()]);
1296 base.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
1297
1298 let inherit = base.with_policy(&PolicyOverrides::default());
1300 assert!(inherit.is_view(), "派生视图 is_view 应为真");
1301 assert!(inherit.require_context());
1302 assert_eq!(inherit.profile(), Profile::Text2Query);
1303 assert!(inherit.rbac().is_some());
1304 assert_eq!(inherit.role_rules().exempt_roles, vec!["a".to_string()]);
1305 assert_eq!(inherit.role_rules().deny_write_roles, vec!["b".to_string()]);
1306 assert_eq!(
1307 inherit.role_rules().unconfigured,
1308 UnconfiguredPolicy::Closed
1309 );
1310 assert!(inherit.meta_policy().closed);
1311 assert_eq!(inherit.meta_policy().roles, vec!["m".to_string()]);
1312 assert_eq!(inherit.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
1313
1314 let full = base.with_policy(&overrides_all());
1316 assert!(!full.require_context());
1317 assert_eq!(full.profile(), Profile::Standard);
1318 assert!(full.rbac().is_none(), "rbac:null 应清除");
1319 assert_eq!(full.role_rules().exempt_roles, vec!["x".to_string()]);
1320 assert_eq!(full.role_rules().deny_write_roles, vec!["y".to_string()]);
1321 assert_eq!(full.role_rules().unconfigured, UnconfiguredPolicy::Open);
1322 assert!(!full.meta_policy().closed);
1323 assert_eq!(full.write_link_policy(), WriteLinkPolicy::Reject);
1324
1325 assert!(base.require_context());
1327 assert_eq!(base.profile(), Profile::Text2Query);
1328 assert!(base.rbac().is_some());
1329 assert!(base.meta_policy().closed);
1330 }
1331
1332 #[test]
1334 fn view_register_is_readonly_err() {
1335 let base = Registry::new();
1336 let mut view = base.with_policy(&PolicyOverrides::default());
1337 let err = view
1338 .register(&json!({ "name": "A", "fields": {} }))
1339 .expect_err("视图注册应 Err");
1340 assert!(
1341 err.starts_with(ERR_POLICY_VIEW_READONLY),
1342 "前缀可匹配: {err}"
1343 );
1344 let err2 = view
1345 .register_with_ctx(&json!({ "name": "A", "fields": {} }), None)
1346 .expect_err("视图注册应 Err");
1347 assert!(
1348 err2.starts_with(ERR_POLICY_VIEW_READONLY),
1349 "前缀可匹配: {err2}"
1350 );
1351 let err3 = view
1352 .register_batch(
1353 &[(json!({ "name": "A", "fields": {} }), Location::default())],
1354 None,
1355 )
1356 .expect_err("视图批量注册应 Err");
1357 assert!(
1358 err3.starts_with(ERR_POLICY_VIEW_READONLY),
1359 "前缀可匹配: {err3}"
1360 );
1361 }
1362
1363 #[test]
1365 #[should_panic(expected = "ERR_POLICY_VIEW_READONLY")]
1366 fn view_clear_panics_readonly() {
1367 let base = Registry::new();
1368 let mut view = base.with_policy(&PolicyOverrides::default());
1369 view.clear();
1370 }
1371
1372 #[test]
1374 fn view_sees_catalog_snapshot() {
1375 let mut base = Registry::new();
1376 base.register(&json!({ "name": "A", "fields": { "x": { "type": "number" } } }))
1377 .expect("base 注册成功");
1378 let view = base.with_policy(&PolicyOverrides::default());
1379 assert!(view.has("A"), "派生视图应见 base 已注册目录");
1380 assert_eq!(view.get("A").expect("视图可取").name, "A");
1381 assert_eq!(view.list(), base.list());
1382
1383 let mut base2 = Registry::new();
1385 base2
1386 .register(&json!({ "name": "A", "fields": {} }))
1387 .unwrap();
1388 let snap = base2.with_policy(&PolicyOverrides::default());
1389 base2
1390 .register(&json!({ "name": "B", "fields": {} }))
1391 .unwrap();
1392 assert!(snap.has("A"), "在途视图仍持其快照");
1393 let fresh = base2.with_policy(&PolicyOverrides::default());
1394 assert!(fresh.has("B"), "新派生视图应见 base 后续注册");
1395 }
1396
1397 #[test]
1399 fn view_policy_set_isolated() {
1400 let base = Registry::new();
1401 let mut view = base.with_policy(&PolicyOverrides::default());
1402 view.set_profile(Profile::Text2Query);
1403 assert_eq!(view.profile(), Profile::Text2Query);
1404 assert_eq!(base.profile(), Profile::Standard, "视图 set 不影响 base");
1405 let sibling = base.with_policy(&PolicyOverrides::default());
1406 assert_eq!(sibling.profile(), Profile::Standard, "不影响兄弟视图");
1407 }
1408
1409 #[test]
1411 fn from_value_unknown_key_and_rbac_null() {
1412 let e = PolicyOverrides::from_value(&json!({ "nope": 1 })).expect_err("未知键应 Err");
1413 assert!(e.to_string().starts_with(ERR_POLICY_VIEW_READONLY), "{e}");
1414 assert!(
1415 PolicyOverrides::from_value(&json!({ "profile": "bogus" })).is_err(),
1416 "非法 profile 应 Err(禁静默回落)"
1417 );
1418 assert!(
1419 PolicyOverrides::from_value(&json!({ "writeLinkPolicy": "bogus" })).is_err(),
1420 "非法 writeLinkPolicy 应 Err"
1421 );
1422 assert!(
1423 PolicyOverrides::from_value(&json!({})).is_ok(),
1424 "空对象合法"
1425 );
1426
1427 let mut base = Registry::new();
1428 base.set_rbac(Some(&json!({}))).unwrap();
1429 let cleared =
1430 base.with_policy(&PolicyOverrides::from_value(&json!({ "rbac": null })).unwrap());
1431 assert!(cleared.rbac().is_none(), "rbac:null 应清除");
1432 let kept = base.with_policy(&PolicyOverrides::from_value(&json!({})).unwrap());
1433 assert!(kept.rbac().is_some(), "缺键应继承");
1434 assert!(base.rbac().is_some(), "base 不受影响");
1435 }
1436}