1use std::collections::HashMap;
4
5use serde_json::{json, Map, Value};
6
7use crate::types::{is_truthy, str_list, AGG_OPS};
8
9use super::definition::{normalize_fields, ComputeDef, FieldDef, Location, RelationDef, Schema};
10
11use crate::command::WriteLinkPolicy;
12
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
19pub enum Profile {
20 #[default]
22 Standard,
23 Text2Query,
25}
26
27impl Profile {
28 pub fn as_str(&self) -> &'static str {
30 match self {
31 Profile::Standard => "standard",
32 Profile::Text2Query => "text2query",
33 }
34 }
35
36 pub fn from_str_or_err(s: &str) -> Result<Profile, String> {
38 match s {
39 "standard" => Ok(Profile::Standard),
40 "text2query" => Ok(Profile::Text2Query),
41 other => Err(format!("未知 profile: {other}(仅 standard / text2query)")),
42 }
43 }
44}
45
46#[derive(Debug, Clone)]
48struct Entry {
49 schema: Schema,
50 links: Vec<Location>,
52 version: u64,
54}
55
56pub(super) struct BatchClass {
58 pub order: Vec<String>,
60 pub primaries: HashMap<String, (Value, Location, String)>,
62 pub replicas: HashMap<String, Vec<(Value, Location)>>,
64}
65
66pub(super) enum BatchError {
68 Parse(String),
70 PrimaryDuplicate { name: String, indices: Vec<usize> },
72 LocationConflict(Box<LocationConflict>),
74}
75
76pub(super) fn classify_batch(items: &[(Value, Location)]) -> Result<BatchClass, BatchError> {
82 let mut order: Vec<String> = Vec::new();
83 let mut primaries: HashMap<String, (Value, Location, String)> = HashMap::new();
84 let mut primary_idx: HashMap<String, usize> = HashMap::new();
85 let mut replicas: HashMap<String, Vec<(Value, Location)>> = HashMap::new();
86
87 for (idx, (defn, loc)) in items.iter().enumerate() {
88 let obj = defn
89 .as_object()
90 .ok_or_else(|| BatchError::Parse("schema 定义必须是对象".to_string()))?;
91 let name = obj
92 .get("name")
93 .and_then(|v| v.as_str())
94 .ok_or_else(|| BatchError::Parse("schema 缺少 name".to_string()))?
95 .to_string();
96 if !order.iter().any(|n| n == &name) {
97 order.push(name.clone());
98 }
99
100 if obj.get("replica").map(is_truthy).unwrap_or(false) {
101 replicas
102 .entry(name)
103 .or_default()
104 .push((defn.clone(), loc.clone()));
105 continue;
106 }
107
108 let collection = obj
109 .get("collection")
110 .and_then(|v| v.as_str())
111 .filter(|s| !s.is_empty())
112 .map(String::from)
113 .unwrap_or_else(|| name.clone());
114 if let Some(first) = primary_idx.get(&name) {
115 return Err(BatchError::PrimaryDuplicate {
117 name,
118 indices: vec![*first, idx],
119 });
120 }
121 primary_idx.insert(name.clone(), idx);
122 primaries.insert(name, (defn.clone(), loc.clone(), collection));
123 }
124
125 let mut quads: Vec<(String, String, Location)> = Vec::new();
127 for name in &order {
128 if let Some((_, loc, coll)) = primaries.get(name) {
129 quads.push((name.clone(), coll.clone(), loc.clone()));
130 }
131 if let Some(rs) = replicas.get(name) {
132 for (defn, loc) in rs {
133 let coll = defn
134 .get("name")
135 .and_then(|v| v.as_str())
136 .unwrap_or(name)
137 .to_string();
138 quads.push((name.clone(), coll, loc.clone()));
139 }
140 }
141 }
142 if let Some(c) = detect_location_conflict(&quads) {
143 return Err(BatchError::LocationConflict(Box::new(c)));
144 }
145
146 Ok(BatchClass {
147 order,
148 primaries,
149 replicas,
150 })
151}
152
153pub(super) struct LocationConflict {
155 pub collection: String,
156 pub prev: String,
157 pub name: String,
158 pub source: String,
159 pub database: Option<String>,
160 pub schema: Option<String>,
161}
162
163pub(super) fn detect_location_conflict(
167 entries: &[(String, String, Location)],
168) -> Option<LocationConflict> {
169 let mut seen: HashMap<(String, Option<String>, Option<String>, String), String> =
170 HashMap::new();
171 for (name, collection, loc) in entries {
172 let key = (
173 loc.source.clone(),
174 loc.database.clone(),
175 loc.schema.clone(),
176 collection.clone(),
177 );
178 if let Some(prev) = seen.get(&key) {
179 if prev != name {
180 return Some(LocationConflict {
181 collection: collection.clone(),
182 prev: prev.clone(),
183 name: name.clone(),
184 source: loc.source.clone(),
185 database: loc.database.clone(),
186 schema: loc.schema.clone(),
187 });
188 }
189 } else {
190 seen.insert(key, name.clone());
191 }
192 }
193 None
194}
195
196fn batch_err_to_msg(e: BatchError) -> String {
198 match e {
199 BatchError::Parse(m) => m,
200 BatchError::PrimaryDuplicate { name, .. } => format!(
201 "同名主定义重复: {}(同一装载批次内 name 必须唯一,从定义请用 replica: true)",
202 name
203 ),
204 BatchError::LocationConflict(c) => format!(
205 "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
206 c.source, c.database, c.schema, c.collection, c.prev
207 ),
208 }
209}
210
211#[derive(Debug, Clone, Default)]
212pub struct Registry {
213 entries: HashMap<String, Entry>,
214 order: Vec<String>,
215 require_context: bool,
219 profile: Profile,
221 rbac: Option<crate::rbac::RbacPolicy>,
223 role_rules: crate::permission::RoleRules,
225 meta_policy: crate::permission::MetaPolicy,
227 write_link_policy: WriteLinkPolicy,
229}
230
231impl Registry {
232 pub fn new() -> Self {
233 Self::default()
234 }
235
236 pub fn register(&mut self, defn: &Value) -> Result<(), String> {
242 self.register_with_ctx(defn, None)
243 }
244
245 pub fn register_with_ctx(
247 &mut self,
248 defn: &Value,
249 ctx: Option<&crate::permission::Context>,
250 ) -> Result<(), String> {
251 self.register_batch(&[(defn.clone(), Location::default())], ctx)
252 }
253
254 pub fn register_batch(
269 &mut self,
270 items: &[(Value, Location)],
271 ctx: Option<&crate::permission::Context>,
272 ) -> Result<(), String> {
273 let class = classify_batch(items).map_err(batch_err_to_msg)?;
275
276 for (defn, _loc) in items {
278 let name = defn.get("name").and_then(|v| v.as_str()).unwrap_or("");
279 if !crate::permission::can_register(&self.meta_policy, ctx) {
280 return Err(format!("ERR_PERMISSION: 无权注册或覆盖定义 {name}"));
281 }
282 }
283
284 struct Planned {
286 name: String,
287 schema: Schema,
288 links: Vec<Location>,
289 }
290 let mut planned: Vec<Planned> = Vec::new();
291 for name in &class.order {
292 let rs: Vec<Location> = class
293 .replicas
294 .get(name)
295 .map(|v| v.iter().map(|(_, l)| l.clone()).collect())
296 .unwrap_or_default();
297 if let Some((raw, loc, collection)) = class.primaries.get(name) {
298 let obj = raw.as_object().unwrap();
299 let schema = build_schema(obj, name.clone(), collection.clone(), loc)?;
300 let is_archive = obj.get("_isArchive").map(is_truthy).unwrap_or(false);
301 let mut links = vec![loc.clone()];
302 links.extend(rs);
303 planned.push(Planned {
304 name: name.clone(),
305 schema: schema.clone(),
306 links: links.clone(),
307 });
308 if !is_archive && !name.ends_with("Deleted") {
310 let aobj = archive_defn(obj, name, &schema.collection)
311 .as_object()
312 .cloned()
313 .unwrap();
314 let aname = format!("{}Deleted", name);
315 let acoll = format!("{}_deleted", schema.collection);
316 let aschema = build_schema(&aobj, aname.clone(), acoll, &links[0])?;
317 planned.push(Planned {
318 name: aname,
319 schema: aschema,
320 links,
321 });
322 }
323 } else {
324 let Some(existing) = self.entries.get(name) else {
326 return Err(format!(
327 "未找到主 schema: {}(从定义 `replica` 必须伴随同批主定义,或该 name 已注册)",
328 name
329 ));
330 };
331 let mut links = existing.links.clone();
332 links.extend(rs);
333 planned.push(Planned {
334 name: name.clone(),
335 schema: existing.schema.clone(),
336 links,
337 });
338 }
339 }
340
341 let batch_names: std::collections::HashSet<&str> =
343 planned.iter().map(|p| p.name.as_str()).collect();
344 let mut quads: Vec<(String, String, Location)> = Vec::new();
345 for (n, e) in &self.entries {
346 if batch_names.contains(n.as_str()) {
347 continue; }
349 for l in &e.links {
350 quads.push((n.clone(), e.schema.collection.clone(), l.clone()));
351 }
352 }
353 for p in &planned {
354 for l in &p.links {
355 quads.push((p.name.clone(), p.schema.collection.clone(), l.clone()));
356 }
357 }
358 if let Some(c) = detect_location_conflict(&quads) {
359 return Err(format!(
360 "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
361 c.source, c.database, c.schema, c.collection, c.prev
362 ));
363 }
364
365 for p in planned {
367 if let Some(e) = self.entries.get_mut(&p.name) {
368 e.schema = p.schema;
369 e.links = p.links;
370 e.version += 1;
371 } else {
372 self.entries.insert(
373 p.name.clone(),
374 Entry {
375 schema: p.schema,
376 links: p.links,
377 version: 1,
378 },
379 );
380 self.order.push(p.name);
381 }
382 }
383
384 Ok(())
385 }
386
387 pub fn links_of(&self, name: &str) -> Result<&[Location], String> {
389 Ok(&self.get_entry(name)?.links)
390 }
391
392 pub fn primary_location(&self, name: &str) -> Result<&Location, String> {
394 self.get_entry(name)?
395 .links
396 .first()
397 .ok_or_else(|| format!("Schema 无落点链路: {}", name))
398 }
399
400 pub fn version_of(&self, name: &str) -> Result<u64, String> {
402 Ok(self.get_entry(name)?.version)
403 }
404
405 fn get_entry(&self, name: &str) -> Result<&Entry, String> {
406 self.entries
407 .get(name)
408 .ok_or_else(|| format!("Schema 未注册: {}", name))
409 }
410
411 pub fn get(&self, name: &str) -> Result<&Schema, String> {
413 self.get_entry(name).map(|e| &e.schema)
414 }
415
416 pub fn has(&self, name: &str) -> bool {
417 self.entries.contains_key(name)
418 }
419
420 pub fn clear(&mut self) {
426 self.entries.clear();
427 self.order.clear();
428 }
429
430 pub fn set_require_context(&mut self, require: bool) {
434 self.require_context = require;
435 }
436
437 pub fn require_context(&self) -> bool {
439 self.require_context
440 }
441
442 pub fn set_profile(&mut self, profile: Profile) {
447 self.profile = profile;
448 }
449
450 pub fn profile(&self) -> Profile {
452 self.profile
453 }
454
455 pub fn set_exempt_roles(&mut self, roles: Vec<String>) {
457 self.role_rules.exempt_roles = roles;
458 }
459
460 pub fn set_deny_write_roles(&mut self, roles: Vec<String>) {
462 self.role_rules.deny_write_roles = roles;
463 }
464
465 pub fn set_unconfigured_policy(&mut self, policy: crate::permission::UnconfiguredPolicy) {
467 self.role_rules.unconfigured = policy;
468 }
469
470 pub fn role_rules(&self) -> &crate::permission::RoleRules {
472 &self.role_rules
473 }
474
475 pub fn set_meta_policy(&mut self, closed: bool, roles: Vec<String>) {
478 self.meta_policy = crate::permission::MetaPolicy { closed, roles };
479 }
480
481 pub fn meta_policy(&self) -> &crate::permission::MetaPolicy {
483 &self.meta_policy
484 }
485
486 pub fn set_write_link_policy(&mut self, policy: WriteLinkPolicy) {
491 self.write_link_policy = policy;
492 }
493
494 pub fn write_link_policy(&self) -> WriteLinkPolicy {
496 self.write_link_policy
497 }
498
499 pub fn can_register(&self, ctx: Option<&crate::permission::Context>) -> bool {
505 crate::permission::can_register(&self.meta_policy, ctx)
506 }
507
508 pub fn set_rbac(&mut self, policy: Option<&Value>) -> Result<(), String> {
511 match policy {
512 None => {
513 self.rbac = None;
514 Ok(())
515 }
516 Some(v) => {
517 let p = crate::rbac::RbacPolicy::from_json(v)?;
518 self.rbac = Some(p);
519 Ok(())
520 }
521 }
522 }
523
524 pub fn rbac(&self) -> Option<&crate::rbac::RbacPolicy> {
526 self.rbac.as_ref()
527 }
528
529 pub fn get_by_location(
534 &self,
535 source: &str,
536 database: Option<&str>,
537 schema: Option<&str>,
538 collection: &str,
539 ) -> Result<&Schema, String> {
540 let db = database.filter(|s| !s.is_empty());
541 let sc = schema.filter(|s| !s.is_empty());
542 self.entries
543 .values()
544 .find(|e| {
545 e.schema.collection == collection
546 && e.links.iter().any(|l| {
547 l.source == source
548 && l.database.as_deref().filter(|s| !s.is_empty()) == db
549 && l.schema.as_deref().filter(|s| !s.is_empty()) == sc
550 })
551 })
552 .map(|e| &e.schema)
553 .ok_or_else(|| {
554 format!(
555 "Schema 未定位(source = {}, database = {:?}, schema = {:?}, collection = {})",
556 source, db, sc, collection
557 )
558 })
559 }
560
561 pub fn get_for_command(
568 &self,
569 source: &str,
570 database: Option<&str>,
571 schema: Option<&str>,
572 collection: &str,
573 ) -> Result<&Schema, String> {
574 if let Ok(s) = self.get_by_location(source, database, schema, collection) {
575 return Ok(s);
576 }
577 let candidates: Vec<&Entry> = self
578 .entries
579 .values()
580 .filter(|e| {
581 e.schema.collection == collection && e.links.iter().any(|l| l.source == source)
582 })
583 .collect();
584 match candidates.len() {
585 0 => self.get_by_location(source, database, schema, collection),
586 1 => Ok(&candidates[0].schema),
587 _ => candidates
588 .iter()
589 .find(|e| e.schema.database().is_none() && e.schema.schema().is_none())
590 .map(|e| &e.schema)
591 .ok_or_else(|| {
592 format!(
593 "命令定位 ({}, {:?}, {:?}, {}) 存在多个结构 schema(多租户 override 需唯一的 (source, collection) 结构声明)",
594 source, database, schema, collection
595 )
596 }),
597 }
598 }
599
600 pub fn list(&self) -> Vec<String> {
601 self.order.clone()
602 }
603
604 pub fn schema_datasource(&self, name: &str) -> Result<Option<String>, String> {
606 Ok(self.get(name)?.source.clone())
607 }
608
609 pub fn resolve_datasource(
614 &self,
615 schema_name: &str,
616 config: &Value,
617 ) -> Result<crate::datasource::DataSource, String> {
618 let declared = self.get(schema_name)?.source.as_deref();
619 let cfg = crate::datasource::DataSourceConfig::from_json(config)?;
620 cfg.resolve(declared)
621 }
622}
623
624fn build_schema(
627 obj: &Map<String, Value>,
628 name: String,
629 collection: String,
630 loc: &Location,
631) -> Result<Schema, String> {
632 let timestamps = parse_timestamps(obj)?;
633 let mut fields = normalize_fields(obj.get("fields"))?;
634 if timestamps {
635 add_timestamp_fields(&mut fields);
636 }
637 let relations = parse_relations(obj);
638 let computes = parse_computes(obj, &relations)?;
640 let triggers = super::triggers::parse_triggers(obj, &name, &fields)?;
642 let mut ids: Vec<&str> = Vec::new();
645 ids.extend(fields.keys().map(String::as_str));
646 ids.extend(relations.keys().map(String::as_str));
647 ids.extend(computes.iter().map(|(k, _)| k.as_str()));
648 crate::naming::detect_conflicts(ids)?;
649 Ok(Schema {
650 name,
651 collection,
652 id_prefix: obj
653 .get("idPrefix")
654 .and_then(|v| v.as_str())
655 .unwrap_or("")
656 .to_string(),
657 timestamps,
658 fields,
659 relations,
660 computes,
661 triggers,
662 read: str_list(obj.get("read")),
663 write: str_list(obj.get("write")),
664 indexes: obj
665 .get("indexes")
666 .and_then(|v| v.as_array())
667 .cloned()
668 .unwrap_or_default(),
669 source: if loc.source == crate::datasource::DEFAULT_SOURCE {
670 None
671 } else {
672 Some(loc.source.clone())
673 },
674 database: loc.database.clone().filter(|s| !s.is_empty()),
675 schema: loc.schema.clone().filter(|s| !s.is_empty()),
676 })
677}
678
679fn parse_timestamps(obj: &Map<String, Value>) -> Result<bool, String> {
681 match obj.get("timestamps") {
682 None | Some(Value::Null) | Some(Value::Bool(true)) => Ok(true),
683 Some(Value::Bool(false)) => Ok(false),
684 Some(Value::String(s)) if s == "ms" || s == "s" => Ok(true),
685 Some(Value::String(s)) => Err(format!(
686 "timestamps 仅支持 true/false/'ms'/'s',实际 {:?}",
687 s
688 )),
689 Some(other) => Err(format!(
690 "timestamps 仅支持 true/false/'ms'/'s',实际 {}",
691 other
692 )),
693 }
694}
695
696fn add_timestamp_fields(fields: &mut HashMap<String, FieldDef>) {
698 for key in ["createdAt", "updatedAt"] {
699 fields.entry(key.to_string()).or_insert_with(|| FieldDef {
700 field_type: "number".to_string(),
701 required: false,
702 default: None,
703 read: None,
704 write: None,
705 fields: None,
706 strategy: None,
707 });
708 }
709}
710
711fn parse_relations(obj: &Map<String, Value>) -> HashMap<String, RelationDef> {
712 let mut relations = HashMap::new();
713 if let Some(Value::Object(rm)) = obj.get("relations") {
714 for (key, val) in rm {
715 let vo = val.as_object();
716 let get_str = |k: &str| -> Option<String> {
717 vo.and_then(|o| o.get(k))
718 .and_then(|v| v.as_str())
719 .filter(|s| !s.is_empty())
720 .map(String::from)
721 };
722 relations.insert(
723 key.clone(),
724 RelationDef {
725 model: get_str("model").unwrap_or_default(),
726 rel_type: get_str("type").unwrap_or_else(|| "many".to_string()),
727 local_field: get_str("localField").unwrap_or_else(|| "_id".to_string()),
728 foreign_field: get_str("foreignField").unwrap_or_else(|| key.clone()),
729 read: str_list(vo.and_then(|o| o.get("read"))),
730 },
731 );
732 }
733 }
734 relations
735}
736
737fn parse_computes(
738 obj: &Map<String, Value>,
739 relations: &HashMap<String, RelationDef>,
740) -> Result<Vec<(String, ComputeDef)>, String> {
741 let mut computes = Vec::new();
742 if let Some(Value::Object(cm)) = obj.get("computes") {
743 for (key, val) in cm {
744 let vo = val.as_object();
745 let get = |k: &str| vo.and_then(|o| o.get(k)).cloned();
746 let has_fn = get("fn").map(|v| is_truthy(&v)).unwrap_or(false);
747 let has_async_fn = get("asyncFn").map(|v| is_truthy(&v)).unwrap_or(false);
748 let agg = parse_agg(key, get("agg").as_ref(), relations, has_fn || has_async_fn)?;
749 computes.push((
750 key.clone(),
751 ComputeDef {
752 comp_type: get("type")
753 .and_then(|v| v.as_str().map(String::from))
754 .filter(|s| !s.is_empty())
755 .unwrap_or_else(|| "any".to_string()),
756 has_fn,
757 has_async_fn,
758 fn_ref: get("fnRef")
759 .and_then(|v| v.as_str().map(String::from))
760 .filter(|s| !s.is_empty()),
761 agg,
762 depends: str_list(get("depends").as_ref()).unwrap_or_default(),
763 read: str_list(get("read").as_ref()),
764 },
765 ));
766 }
767 }
768 Ok(computes)
769}
770
771fn parse_agg(
779 key: &str,
780 raw: Option<&Value>,
781 relations: &HashMap<String, RelationDef>,
782 has_callback: bool,
783) -> Result<Option<Value>, String> {
784 let Some(v) = raw.filter(|v| is_truthy(v)) else {
785 return Ok(None);
786 };
787 if has_callback {
788 return Err(format!(
789 "计算列 \"{key}\" 不能同时声明 agg 与 fn/asyncFn(二选一)"
790 ));
791 }
792 let obj = v.as_object().ok_or_else(|| {
793 format!("计算列 \"{key}\" 的 agg 必须是对象,如 {{\"$count\": \"lessons\"}}")
794 })?;
795 let mut it = obj.iter();
796 let (op, arg) = match (it.next(), it.next()) {
797 (Some(kv), None) => kv,
798 _ => return Err(format!("计算列 \"{key}\" 的 agg 必须且只能声明一个算子键")),
799 };
800 if !AGG_OPS.contains(&op.as_str()) {
801 return Err(format!(
802 "计算列 \"{key}\" 的 agg 算子 {op} 不在白名单($count/$sum/$avg/$min/$max)"
803 ));
804 }
805 let path = arg
806 .as_str()
807 .filter(|s| !s.is_empty() && *s != "*")
808 .ok_or_else(|| {
809 format!("计算列 \"{key}\" 的 agg 取值必须是关系路径字符串(如 \"lessons\" / \"lessons.duration\")")
810 })?;
811 let (rel, field) = match path.split_once('.') {
812 Some((r, f)) => (r, Some(f)),
813 None => (path, None),
814 };
815 if !relations.contains_key(rel) {
816 return Err(format!(
817 "计算列 \"{key}\" 的 agg 引用的关系 \"{rel}\" 未在 schema 中定义"
818 ));
819 }
820 if field
821 .map(|f| f.is_empty() || f.contains('.'))
822 .unwrap_or(false)
823 {
824 return Err(format!(
825 "计算列 \"{key}\" 的 agg 关系路径 \"{path}\" 非法:仅支持单级「关系.字段」(二级路径首批不支持)"
826 ));
827 }
828 match op.as_str() {
829 "$count" if field.is_some() => Err(format!(
830 "计算列 \"{key}\" 的 $count 仅支持关系整行计数(如 \"lessons\"),按字段计数待 P5"
831 )),
832 "$count" => Ok(Some(v.clone())),
833 _ if field.is_none() => Err(format!(
834 "计算列 \"{key}\" 的 agg 算子 {op} 必须引用关系字段(如 \"lessons.duration\")"
835 )),
836 _ => Ok(Some(v.clone())),
837 }
838}
839
840fn archive_defn(obj: &Map<String, Value>, name: &str, collection: &str) -> Value {
843 let mut arch_fields = obj
844 .get("fields")
845 .and_then(|v| v.as_object())
846 .cloned()
847 .unwrap_or_default();
848 arch_fields.insert("deletedAt".to_string(), json!({ "type": "number" }));
849 if let Some(idf) = arch_fields.get_mut("_id") {
852 if let Some(o) = idf.as_object_mut() {
853 o.remove("strategy");
854 }
855 }
856 let arch = json!({
857 "name": format!("{}Deleted", name),
858 "collection": format!("{}_deleted", collection),
859 "idPrefix": "",
860 "_isArchive": true,
861 "fields": Value::Object(arch_fields),
862 "indexes": obj.get("indexes").cloned().unwrap_or_else(|| json!([])),
863 });
864 arch
865}
866
867#[cfg(test)]
868mod tests {
869 use super::*;
870
871 #[test]
872 fn write_link_policy_defaults_to_reject() {
873 let r = Registry::new();
874 assert_eq!(r.write_link_policy(), WriteLinkPolicy::Reject);
875 let mut r2 = Registry::new();
876 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::Reject);
877 r2.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
878 assert_eq!(r2.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
879 }
880}