use std::collections::HashMap;
use std::sync::Arc;
use serde_json::{json, Map, Value};
use crate::types::{is_truthy, str_list, AGG_OPS};
use super::definition::{
normalize_fields, validate_vector_fields, ComputeDef, FieldDef, Location, RelationDef, Schema,
};
use super::load::LoadConfig;
use crate::command::WriteLinkPolicy;
use crate::error::{CoreError, ERR_POLICY_VIEW_READONLY};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Profile {
#[default]
Standard,
Text2Query,
}
impl Profile {
pub fn as_str(&self) -> &'static str {
match self {
Profile::Standard => "standard",
Profile::Text2Query => "text2query",
}
}
pub fn from_str_or_err(s: &str) -> Result<Profile, String> {
match s {
"standard" => Ok(Profile::Standard),
"text2query" => Ok(Profile::Text2Query),
other => Err(format!("未知 profile: {other}(仅 standard / text2query)")),
}
}
}
#[derive(Debug, Clone)]
struct Entry {
schema: Schema,
links: Vec<Location>,
version: u64,
}
pub(super) struct BatchClass {
pub order: Vec<String>,
pub primaries: HashMap<String, (Value, Location, String)>,
pub replicas: HashMap<String, Vec<(Value, Location)>>,
}
pub(super) enum BatchError {
Parse(String),
PrimaryDuplicate { name: String, indices: Vec<usize> },
LocationConflict(Box<LocationConflict>),
}
pub(super) fn classify_batch(items: &[(Value, Location)]) -> Result<BatchClass, BatchError> {
let mut order: Vec<String> = Vec::new();
let mut primaries: HashMap<String, (Value, Location, String)> = HashMap::new();
let mut primary_idx: HashMap<String, usize> = HashMap::new();
let mut replicas: HashMap<String, Vec<(Value, Location)>> = HashMap::new();
for (idx, (defn, loc)) in items.iter().enumerate() {
let obj = defn
.as_object()
.ok_or_else(|| BatchError::Parse("schema 定义必须是对象".to_string()))?;
let name = obj
.get("name")
.and_then(|v| v.as_str())
.ok_or_else(|| BatchError::Parse("schema 缺少 name".to_string()))?
.to_string();
if !order.iter().any(|n| n == &name) {
order.push(name.clone());
}
if obj.get("replica").map(is_truthy).unwrap_or(false) {
replicas
.entry(name)
.or_default()
.push((defn.clone(), loc.clone()));
continue;
}
let collection = obj
.get("collection")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.map(String::from)
.unwrap_or_else(|| name.clone());
if let Some(first) = primary_idx.get(&name) {
return Err(BatchError::PrimaryDuplicate {
name,
indices: vec![*first, idx],
});
}
primary_idx.insert(name.clone(), idx);
primaries.insert(name, (defn.clone(), loc.clone(), collection));
}
let mut quads: Vec<(String, String, Location)> = Vec::new();
for name in &order {
if let Some((_, loc, coll)) = primaries.get(name) {
quads.push((name.clone(), coll.clone(), loc.clone()));
}
if let Some(rs) = replicas.get(name) {
for (defn, loc) in rs {
let coll = defn
.get("name")
.and_then(|v| v.as_str())
.unwrap_or(name)
.to_string();
quads.push((name.clone(), coll, loc.clone()));
}
}
}
if let Some(c) = detect_location_conflict(&quads) {
return Err(BatchError::LocationConflict(Box::new(c)));
}
Ok(BatchClass {
order,
primaries,
replicas,
})
}
pub(super) struct LocationConflict {
pub collection: String,
pub prev: String,
pub name: String,
pub source: String,
pub database: Option<String>,
pub schema: Option<String>,
}
pub(super) fn detect_location_conflict(
entries: &[(String, String, Location)],
) -> Option<LocationConflict> {
let mut seen: HashMap<(String, Option<String>, Option<String>, String), String> =
HashMap::new();
for (name, collection, loc) in entries {
let key = (
loc.source.clone(),
loc.database.clone(),
loc.schema.clone(),
collection.clone(),
);
if let Some(prev) = seen.get(&key) {
if prev != name {
return Some(LocationConflict {
collection: collection.clone(),
prev: prev.clone(),
name: name.clone(),
source: loc.source.clone(),
database: loc.database.clone(),
schema: loc.schema.clone(),
});
}
} else {
seen.insert(key, name.clone());
}
}
None
}
fn batch_err_to_msg(e: BatchError) -> String {
match e {
BatchError::Parse(m) => m,
BatchError::PrimaryDuplicate { name, .. } => format!(
"同名主定义重复: {}(同一装载批次内 name 必须唯一,从定义请用 replica: true)",
name
),
BatchError::LocationConflict(c) => format!(
"定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
c.source, c.database, c.schema, c.collection, c.prev
),
}
}
#[derive(Debug, Clone, Default)]
pub struct Catalog {
entries: HashMap<String, Entry>,
order: Vec<String>,
}
#[derive(Debug, Clone, Default)]
pub struct PolicyBundle {
pub(crate) require_context: bool,
pub(crate) profile: Profile,
pub(crate) rbac: Option<crate::rbac::RbacPolicy>,
pub(crate) role_rules: crate::permission::RoleRules,
pub(crate) meta_policy: crate::permission::MetaPolicy,
pub(crate) write_link_policy: WriteLinkPolicy,
}
#[derive(Debug, Clone)]
pub struct Registry {
catalog: Arc<Catalog>,
policy: PolicyBundle,
is_base: bool,
}
impl Default for Registry {
fn default() -> Self {
Self {
catalog: Arc::new(Catalog::default()),
policy: PolicyBundle::default(),
is_base: true,
}
}
}
#[derive(Debug, Clone, Default)]
pub struct PolicyOverrides {
require_context: Option<bool>,
profile: Option<Profile>,
rbac: Option<Option<crate::rbac::RbacPolicy>>,
role_rules: Option<crate::permission::RoleRules>,
meta_policy: Option<crate::permission::MetaPolicy>,
write_link_policy: Option<WriteLinkPolicy>,
}
impl PolicyOverrides {
pub fn from_value(v: &Value) -> Result<Self, CoreError> {
let obj = v
.as_object()
.ok_or_else(|| CoreError::Other("策略覆盖必须是 JSON 对象".to_string()))?;
let mut o = PolicyOverrides::default();
for (k, val) in obj {
match k.as_str() {
"profile" => {
let s = val
.as_str()
.ok_or_else(|| CoreError::Other("profile 必须是字符串".to_string()))?;
o.profile = Some(Profile::from_str_or_err(s).map_err(CoreError::Other)?);
}
"requireContext" => {
o.require_context = Some(val.as_bool().ok_or_else(|| {
CoreError::Other("requireContext 必须是布尔".to_string())
})?);
}
"rbac" => {
o.rbac = Some(match val {
Value::Null => None,
other => Some(
crate::rbac::RbacPolicy::from_json(other).map_err(CoreError::Other)?,
),
});
}
"roleRules" => {
o.role_rules = Some(parse_role_rules(val)?);
}
"metaPolicy" => {
o.meta_policy = Some(parse_meta_policy(val)?);
}
"writeLinkPolicy" => {
let s = val.as_str().ok_or_else(|| {
CoreError::Other("writeLinkPolicy 必须是字符串".to_string())
})?;
o.write_link_policy = Some(match s {
"reject" => WriteLinkPolicy::Reject,
"primaryOnly" => WriteLinkPolicy::PrimaryOnly,
other => {
return Err(CoreError::Other(format!(
"未知 writeLinkPolicy: {other}(仅 reject / primaryOnly)"
)))
}
});
}
other => {
return Err(CoreError::Other(format!(
"{ERR_POLICY_VIEW_READONLY} 未知策略覆盖键 `{other}`"
)))
}
}
}
Ok(o)
}
}
fn parse_role_rules(v: &Value) -> Result<crate::permission::RoleRules, CoreError> {
let obj = v
.as_object()
.ok_or_else(|| CoreError::Other("roleRules 必须是 JSON 对象".to_string()))?;
let mut rr = crate::permission::RoleRules::default();
for (k, val) in obj {
match k.as_str() {
"exemptRoles" => rr.exempt_roles = str_list_strict(val, "roleRules.exemptRoles")?,
"denyWriteRoles" => {
rr.deny_write_roles = str_list_strict(val, "roleRules.denyWriteRoles")?
}
"unconfigured" => {
let s = val.as_str().ok_or_else(|| {
CoreError::Other("roleRules.unconfigured 必须是字符串".to_string())
})?;
rr.unconfigured = crate::permission::UnconfiguredPolicy::from_str_or_err(s)
.map_err(CoreError::Other)?;
}
other => {
return Err(CoreError::Other(format!(
"{ERR_POLICY_VIEW_READONLY} 未知 roleRules 键 `{other}`"
)))
}
}
}
Ok(rr)
}
fn parse_meta_policy(v: &Value) -> Result<crate::permission::MetaPolicy, CoreError> {
let obj = v
.as_object()
.ok_or_else(|| CoreError::Other("metaPolicy 必须是 JSON 对象".to_string()))?;
let mut mp = crate::permission::MetaPolicy::default();
for (k, val) in obj {
match k.as_str() {
"closed" => {
mp.closed = val
.as_bool()
.ok_or_else(|| CoreError::Other("metaPolicy.closed 必须是布尔".to_string()))?;
}
"roles" => mp.roles = str_list_strict(val, "metaPolicy.roles")?,
other => {
return Err(CoreError::Other(format!(
"{ERR_POLICY_VIEW_READONLY} 未知 metaPolicy 键 `{other}`"
)))
}
}
}
Ok(mp)
}
fn str_list_strict(v: &Value, key: &str) -> Result<Vec<String>, CoreError> {
let arr = v
.as_array()
.ok_or_else(|| CoreError::Other(format!("{key} 必须是字符串数组")))?;
arr.iter()
.map(|x| {
x.as_str()
.map(String::from)
.ok_or_else(|| CoreError::Other(format!("{key} 的元素必须都是字符串")))
})
.collect()
}
impl PolicyBundle {
fn overridden(&self, o: &PolicyOverrides) -> PolicyBundle {
PolicyBundle {
require_context: o.require_context.unwrap_or(self.require_context),
profile: o.profile.unwrap_or(self.profile),
rbac: match &o.rbac {
Some(v) => v.clone(),
None => self.rbac.clone(),
},
role_rules: o
.role_rules
.clone()
.unwrap_or_else(|| self.role_rules.clone()),
meta_policy: o
.meta_policy
.clone()
.unwrap_or_else(|| self.meta_policy.clone()),
write_link_policy: o.write_link_policy.unwrap_or(self.write_link_policy),
}
}
}
#[derive(Debug, Clone)]
pub struct ReadLinks {
pub links: Vec<Location>,
pub degraded: Vec<Value>,
}
impl Registry {
pub fn new() -> Self {
Self::default()
}
pub fn with_policy(&self, overrides: &PolicyOverrides) -> Registry {
Registry {
catalog: Arc::clone(&self.catalog),
policy: self.policy.overridden(overrides),
is_base: false,
}
}
pub fn is_view(&self) -> bool {
!self.is_base
}
fn ensure_base_catalog(&self) -> Result<(), String> {
if !self.is_base {
return Err(format!(
"{ERR_POLICY_VIEW_READONLY} 非 base 策略视图禁止注册/清空 schema(目录唯一真源在 base Registry)"
));
}
Ok(())
}
pub fn register(&mut self, defn: &Value) -> Result<(), String> {
self.register_with_ctx(defn, None)
}
pub fn register_with_ctx(
&mut self,
defn: &Value,
ctx: Option<&crate::permission::Context>,
) -> Result<(), String> {
self.register_batch(&[(defn.clone(), Location::default())], ctx)
}
pub fn register_batch(
&mut self,
items: &[(Value, Location)],
ctx: Option<&crate::permission::Context>,
) -> Result<(), String> {
self.ensure_base_catalog()?; let class = classify_batch(items).map_err(batch_err_to_msg)?;
for (defn, _loc) in items {
let name = defn.get("name").and_then(|v| v.as_str()).unwrap_or("");
if !crate::permission::can_register(&self.policy.meta_policy, ctx) {
return Err(format!("ERR_PERMISSION: 无权注册或覆盖定义 {name}"));
}
}
struct Planned {
name: String,
schema: Schema,
links: Vec<Location>,
}
let mut planned: Vec<Planned> = Vec::new();
for name in &class.order {
let rs: Vec<Location> = class
.replicas
.get(name)
.map(|v| v.iter().map(|(_, l)| l.clone()).collect())
.unwrap_or_default();
if let Some((raw, loc, collection)) = class.primaries.get(name) {
let obj = raw.as_object().unwrap();
let schema = build_schema(obj, name.clone(), collection.clone(), loc)?;
let is_archive = obj.get("_isArchive").map(is_truthy).unwrap_or(false);
let mut links = vec![loc.clone()];
links.extend(rs);
planned.push(Planned {
name: name.clone(),
schema: schema.clone(),
links: links.clone(),
});
if !is_archive && !name.ends_with("Deleted") {
let aobj = archive_defn(obj, name, &schema.collection)
.as_object()
.cloned()
.unwrap();
let aname = format!("{}Deleted", name);
let acoll = format!("{}_deleted", schema.collection);
let aschema = build_schema(&aobj, aname.clone(), acoll, &links[0])?;
planned.push(Planned {
name: aname,
schema: aschema,
links,
});
}
} else {
let Some(existing) = self.catalog.entries.get(name) else {
return Err(format!(
"未找到主 schema: {}(从定义 `replica` 必须伴随同批主定义,或该 name 已注册)",
name
));
};
let mut links = existing.links.clone();
links.extend(rs);
planned.push(Planned {
name: name.clone(),
schema: existing.schema.clone(),
links,
});
}
}
let batch_names: std::collections::HashSet<&str> =
planned.iter().map(|p| p.name.as_str()).collect();
let mut quads: Vec<(String, String, Location)> = Vec::new();
for (n, e) in &self.catalog.entries {
if batch_names.contains(n.as_str()) {
continue; }
for l in &e.links {
quads.push((n.clone(), e.schema.collection.clone(), l.clone()));
}
}
for p in &planned {
for l in &p.links {
quads.push((p.name.clone(), p.schema.collection.clone(), l.clone()));
}
}
if let Some(c) = detect_location_conflict(&quads) {
return Err(format!(
"定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
c.source, c.database, c.schema, c.collection, c.prev
));
}
let catalog = Arc::make_mut(&mut self.catalog); for p in planned {
if let Some(e) = catalog.entries.get_mut(&p.name) {
e.schema = p.schema;
e.links = p.links;
e.version += 1;
} else {
catalog.entries.insert(
p.name.clone(),
Entry {
schema: p.schema,
links: p.links,
version: 1,
},
);
catalog.order.push(p.name);
}
}
Ok(())
}
pub fn apply_link_decls(&mut self, cfg: &LoadConfig) -> Result<(), String> {
if cfg.links.is_empty() {
return Ok(());
}
self.ensure_base_catalog()?;
let declared: std::collections::HashSet<&str> =
cfg.links.iter().map(|(n, _)| n.as_str()).collect();
let mut planned: Vec<(String, Vec<Location>)> = Vec::new();
for (name, decls) in &cfg.links {
let entry = self.get_entry(name).map_err(|_| {
format!("未注册 schema: {name}(config.links 引用了不存在的 schema)")
})?;
let mut links = entry.links.clone();
for decl in decls {
let target = Location {
source: decl.source.clone(),
database: decl.database.clone(),
schema: decl.schema.clone(),
role: decl.role,
priority: decl.priority,
};
match links.iter_mut().find(|l| l.same_place(&target)) {
Some(hit) => {
hit.role = decl.role;
hit.priority = decl.priority;
}
None => links.push(target),
}
}
if !name.ends_with("Deleted") {
let aname = format!("{}Deleted", name);
if !declared.contains(aname.as_str()) && self.catalog.entries.contains_key(&aname) {
planned.push((aname, links.clone()));
}
}
planned.push((name.clone(), links));
}
let catalog = Arc::make_mut(&mut self.catalog);
for (name, links) in planned {
if let Some(e) = catalog.entries.get_mut(&name) {
e.links = links;
}
}
Ok(())
}
pub fn links_of(&self, name: &str) -> Result<&[Location], String> {
Ok(&self.get_entry(name)?.links)
}
pub fn resolve_read_links(
&self,
name: &str,
unreachable: Option<&[String]>,
) -> Result<ReadLinks, String> {
let mut links = self.links_of(name)?.to_vec();
links.sort_by_key(|l| {
(
l.role.map_or(4u8, |r| r.rank()),
l.priority.unwrap_or(i64::MAX),
)
});
let mut degraded: Vec<Value> = Vec::new();
if let Some(unreachable) = unreachable {
links.retain(|l| {
if unreachable.iter().any(|s| s == &l.source) {
degraded.push(json!({
"code": "readLinkUnreachable",
"layer": "schema",
"message": format!("schema {} 的链路 {} 不可达,已跳过", name, l.source),
"hint": "检查该数据源连接;或调整 store.config.json 的 links 优先级",
"schema": name,
"source": l.source,
}));
false
} else {
true
}
});
}
if links.is_empty() {
return Err(format!(
"ERR_READ_NO_REACHABLE_LINK: {} 的全部链路不可达(已按 unreachable 跳过)",
name
));
}
Ok(ReadLinks { links, degraded })
}
pub fn primary_location(&self, name: &str) -> Result<&Location, String> {
self.get_entry(name)?
.links
.first()
.ok_or_else(|| format!("Schema 无落点链路: {}", name))
}
pub fn version_of(&self, name: &str) -> Result<u64, String> {
Ok(self.get_entry(name)?.version)
}
fn get_entry(&self, name: &str) -> Result<&Entry, String> {
self.catalog
.entries
.get(name)
.ok_or_else(|| format!("Schema 未注册: {}", name))
}
pub fn get(&self, name: &str) -> Result<&Schema, String> {
self.get_entry(name).map(|e| &e.schema)
}
pub fn has(&self, name: &str) -> bool {
self.catalog.entries.contains_key(name)
}
pub fn clear(&mut self) {
if let Err(e) = self.ensure_base_catalog() {
panic!("{e}");
}
let catalog = Arc::make_mut(&mut self.catalog);
catalog.entries.clear();
catalog.order.clear();
}
pub fn set_require_context(&mut self, require: bool) {
self.policy.require_context = require;
}
pub fn require_context(&self) -> bool {
self.policy.require_context
}
pub fn set_profile(&mut self, profile: Profile) {
self.policy.profile = profile;
}
pub fn profile(&self) -> Profile {
self.policy.profile
}
pub fn set_exempt_roles(&mut self, roles: Vec<String>) {
self.policy.role_rules.exempt_roles = roles;
}
pub fn set_deny_write_roles(&mut self, roles: Vec<String>) {
self.policy.role_rules.deny_write_roles = roles;
}
pub fn set_unconfigured_policy(&mut self, policy: crate::permission::UnconfiguredPolicy) {
self.policy.role_rules.unconfigured = policy;
}
pub fn role_rules(&self) -> &crate::permission::RoleRules {
&self.policy.role_rules
}
pub fn set_meta_policy(&mut self, closed: bool, roles: Vec<String>) {
self.policy.meta_policy = crate::permission::MetaPolicy { closed, roles };
}
pub fn meta_policy(&self) -> &crate::permission::MetaPolicy {
&self.policy.meta_policy
}
pub fn set_write_link_policy(&mut self, policy: WriteLinkPolicy) {
self.policy.write_link_policy = policy;
}
pub fn write_link_policy(&self) -> WriteLinkPolicy {
self.policy.write_link_policy
}
pub fn can_register(&self, ctx: Option<&crate::permission::Context>) -> bool {
crate::permission::can_register(&self.policy.meta_policy, ctx)
}
pub fn set_rbac(&mut self, policy: Option<&Value>) -> Result<(), String> {
match policy {
None => {
self.policy.rbac = None;
Ok(())
}
Some(v) => {
let p = crate::rbac::RbacPolicy::from_json(v)?;
self.policy.rbac = Some(p);
Ok(())
}
}
}
pub fn rbac(&self) -> Option<&crate::rbac::RbacPolicy> {
self.policy.rbac.as_ref()
}
pub fn get_by_location(
&self,
source: &str,
database: Option<&str>,
schema: Option<&str>,
collection: &str,
) -> Result<&Schema, String> {
let db = database.filter(|s| !s.is_empty());
let sc = schema.filter(|s| !s.is_empty());
self.catalog
.entries
.values()
.find(|e| {
e.schema.collection == collection
&& e.links.iter().any(|l| {
l.source == source
&& l.database.as_deref().filter(|s| !s.is_empty()) == db
&& l.schema.as_deref().filter(|s| !s.is_empty()) == sc
})
})
.map(|e| &e.schema)
.ok_or_else(|| {
format!(
"Schema 未定位(source = {}, database = {:?}, schema = {:?}, collection = {})",
source, db, sc, collection
)
})
}
pub fn get_for_command(
&self,
source: &str,
database: Option<&str>,
schema: Option<&str>,
collection: &str,
) -> Result<&Schema, String> {
if let Ok(s) = self.get_by_location(source, database, schema, collection) {
return Ok(s);
}
let candidates: Vec<&Entry> = self
.catalog
.entries
.values()
.filter(|e| {
e.schema.collection == collection && e.links.iter().any(|l| l.source == source)
})
.collect();
match candidates.len() {
0 => self.get_by_location(source, database, schema, collection),
1 => Ok(&candidates[0].schema),
_ => candidates
.iter()
.find(|e| e.schema.database().is_none() && e.schema.schema().is_none())
.map(|e| &e.schema)
.ok_or_else(|| {
format!(
"命令定位 ({}, {:?}, {:?}, {}) 存在多个结构 schema(多租户 override 需唯一的 (source, collection) 结构声明)",
source, database, schema, collection
)
}),
}
}
pub fn list(&self) -> Vec<String> {
self.catalog.order.clone()
}
pub fn schema_datasource(&self, name: &str) -> Result<Option<String>, String> {
Ok(self.get(name)?.source.clone())
}
pub fn resolve_datasource(
&self,
schema_name: &str,
config: &Value,
) -> Result<crate::datasource::DataSource, String> {
let declared = self.get(schema_name)?.source.as_deref();
let cfg = crate::datasource::DataSourceConfig::from_json(config)?;
cfg.resolve(declared)
}
}
fn build_schema(
obj: &Map<String, Value>,
name: String,
collection: String,
loc: &Location,
) -> Result<Schema, String> {
let timestamps = parse_timestamps(obj)?;
let mut fields = normalize_fields(obj.get("fields"))?;
validate_vector_fields(&fields)?;
if timestamps {
add_timestamp_fields(&mut fields);
}
let relations = parse_relations(obj);
let computes = parse_computes(obj, &relations)?;
let triggers = super::triggers::parse_triggers(obj, &name, &fields)?;
for (k, ns) in fields
.keys()
.map(|k| (k, "fields"))
.chain(relations.keys().map(|k| (k, "relations")))
.chain(computes.iter().map(|(k, _)| (k, "computes")))
{
if k == "_score" {
return Err(format!(
"{}名称 \"_score\" 为向量检索评分保留键,不得在 schema {ns} 中声明",
crate::naming::ERR_NAME_CONFLICT
));
}
}
let mut ids: Vec<&str> = Vec::new();
ids.extend(fields.keys().map(String::as_str));
ids.extend(relations.keys().map(String::as_str));
ids.extend(computes.iter().map(|(k, _)| k.as_str()));
crate::naming::detect_conflicts(ids)?;
Ok(Schema {
name,
collection,
id_prefix: obj
.get("idPrefix")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string(),
timestamps,
fields,
relations,
computes,
triggers,
read: str_list(obj.get("read")),
write: str_list(obj.get("write")),
indexes: obj
.get("indexes")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default(),
source: if loc.source == crate::datasource::DEFAULT_SOURCE {
None
} else {
Some(loc.source.clone())
},
database: loc.database.clone().filter(|s| !s.is_empty()),
schema: loc.schema.clone().filter(|s| !s.is_empty()),
})
}
fn parse_timestamps(obj: &Map<String, Value>) -> Result<bool, String> {
match obj.get("timestamps") {
None | Some(Value::Null) | Some(Value::Bool(true)) => Ok(true),
Some(Value::Bool(false)) => Ok(false),
Some(Value::String(s)) if s == "ms" || s == "s" => Ok(true),
Some(Value::String(s)) => Err(format!(
"timestamps 仅支持 true/false/'ms'/'s',实际 {:?}",
s
)),
Some(other) => Err(format!(
"timestamps 仅支持 true/false/'ms'/'s',实际 {}",
other
)),
}
}
fn add_timestamp_fields(fields: &mut HashMap<String, FieldDef>) {
for key in ["createdAt", "updatedAt"] {
fields.entry(key.to_string()).or_insert_with(|| FieldDef {
field_type: "number".to_string(),
required: false,
default: None,
read: None,
write: None,
fields: None,
strategy: None,
dim: None,
metric: None,
});
}
}
fn parse_relations(obj: &Map<String, Value>) -> HashMap<String, RelationDef> {
let mut relations = HashMap::new();
if let Some(Value::Object(rm)) = obj.get("relations") {
for (key, val) in rm {
let vo = val.as_object();
let get_str = |k: &str| -> Option<String> {
vo.and_then(|o| o.get(k))
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.map(String::from)
};
relations.insert(
key.clone(),
RelationDef {
model: get_str("model").unwrap_or_default(),
rel_type: get_str("type").unwrap_or_else(|| "many".to_string()),
local_field: get_str("localField").unwrap_or_else(|| "_id".to_string()),
foreign_field: get_str("foreignField").unwrap_or_else(|| key.clone()),
read: str_list(vo.and_then(|o| o.get("read"))),
},
);
}
}
relations
}
fn parse_computes(
obj: &Map<String, Value>,
relations: &HashMap<String, RelationDef>,
) -> Result<Vec<(String, ComputeDef)>, String> {
let mut computes = Vec::new();
if let Some(Value::Object(cm)) = obj.get("computes") {
for (key, val) in cm {
let vo = val.as_object();
let get = |k: &str| vo.and_then(|o| o.get(k)).cloned();
let has_fn = get("fn").map(|v| is_truthy(&v)).unwrap_or(false);
let has_async_fn = get("asyncFn").map(|v| is_truthy(&v)).unwrap_or(false);
let agg = parse_agg(key, get("agg").as_ref(), relations, has_fn || has_async_fn)?;
computes.push((
key.clone(),
ComputeDef {
comp_type: get("type")
.and_then(|v| v.as_str().map(String::from))
.filter(|s| !s.is_empty())
.unwrap_or_else(|| "any".to_string()),
has_fn,
has_async_fn,
fn_ref: get("fnRef")
.and_then(|v| v.as_str().map(String::from))
.filter(|s| !s.is_empty()),
agg,
depends: str_list(get("depends").as_ref()).unwrap_or_default(),
read: str_list(get("read").as_ref()),
},
));
}
}
Ok(computes)
}
fn parse_agg(
key: &str,
raw: Option<&Value>,
relations: &HashMap<String, RelationDef>,
has_callback: bool,
) -> Result<Option<Value>, String> {
let Some(v) = raw.filter(|v| is_truthy(v)) else {
return Ok(None);
};
if has_callback {
return Err(format!(
"计算列 \"{key}\" 不能同时声明 agg 与 fn/asyncFn(二选一)"
));
}
let obj = v.as_object().ok_or_else(|| {
format!("计算列 \"{key}\" 的 agg 必须是对象,如 {{\"$count\": \"lessons\"}}")
})?;
let mut it = obj.iter();
let (op, arg) = match (it.next(), it.next()) {
(Some(kv), None) => kv,
_ => return Err(format!("计算列 \"{key}\" 的 agg 必须且只能声明一个算子键")),
};
if !AGG_OPS.contains(&op.as_str()) {
return Err(format!(
"计算列 \"{key}\" 的 agg 算子 {op} 不在白名单($count/$sum/$avg/$min/$max)"
));
}
let path = arg
.as_str()
.filter(|s| !s.is_empty() && *s != "*")
.ok_or_else(|| {
format!("计算列 \"{key}\" 的 agg 取值必须是关系路径字符串(如 \"lessons\" / \"lessons.duration\")")
})?;
let (rel, field) = match path.split_once('.') {
Some((r, f)) => (r, Some(f)),
None => (path, None),
};
if !relations.contains_key(rel) {
return Err(format!(
"计算列 \"{key}\" 的 agg 引用的关系 \"{rel}\" 未在 schema 中定义"
));
}
if field
.map(|f| f.is_empty() || f.contains('.'))
.unwrap_or(false)
{
return Err(format!(
"计算列 \"{key}\" 的 agg 关系路径 \"{path}\" 非法:仅支持单级「关系.字段」(二级路径首批不支持)"
));
}
match op.as_str() {
"$count" if field.is_some() => Err(format!(
"计算列 \"{key}\" 的 $count 仅支持关系整行计数(如 \"lessons\"),按字段计数待 P5"
)),
"$count" => Ok(Some(v.clone())),
_ if field.is_none() => Err(format!(
"计算列 \"{key}\" 的 agg 算子 {op} 必须引用关系字段(如 \"lessons.duration\")"
)),
_ => Ok(Some(v.clone())),
}
}
fn archive_defn(obj: &Map<String, Value>, name: &str, collection: &str) -> Value {
let mut arch_fields = obj
.get("fields")
.and_then(|v| v.as_object())
.cloned()
.unwrap_or_default();
arch_fields.insert("deletedAt".to_string(), json!({ "type": "number" }));
if let Some(idf) = arch_fields.get_mut("_id") {
if let Some(o) = idf.as_object_mut() {
o.remove("strategy");
}
}
let arch = json!({
"name": format!("{}Deleted", name),
"collection": format!("{}_deleted", collection),
"idPrefix": "",
"_isArchive": true,
"fields": Value::Object(arch_fields),
"indexes": obj.get("indexes").cloned().unwrap_or_else(|| json!([])),
});
arch
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn write_link_policy_defaults_to_reject() {
let r = Registry::new();
assert_eq!(r.write_link_policy(), WriteLinkPolicy::Reject);
let mut r2 = Registry::new();
assert_eq!(r2.write_link_policy(), WriteLinkPolicy::Reject);
r2.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
assert_eq!(r2.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
}
use crate::permission::UnconfiguredPolicy;
fn overrides_all() -> PolicyOverrides {
PolicyOverrides::from_value(&json!({
"profile": "standard",
"requireContext": false,
"rbac": null,
"roleRules": { "exemptRoles": ["x"], "denyWriteRoles": ["y"], "unconfigured": "open" },
"metaPolicy": { "closed": false, "roles": [] },
"writeLinkPolicy": "reject"
}))
.expect("合法覆盖 JSON")
}
#[test]
fn with_policy_override_matrix() {
let mut base = Registry::new();
base.set_require_context(true);
base.set_profile(Profile::Text2Query);
base.set_rbac(Some(&json!({}))).expect("合法 RBAC");
base.set_exempt_roles(vec!["a".to_string()]);
base.set_deny_write_roles(vec!["b".to_string()]);
base.set_unconfigured_policy(UnconfiguredPolicy::Closed);
base.set_meta_policy(true, vec!["m".to_string()]);
base.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
let inherit = base.with_policy(&PolicyOverrides::default());
assert!(inherit.is_view(), "派生视图 is_view 应为真");
assert!(inherit.require_context());
assert_eq!(inherit.profile(), Profile::Text2Query);
assert!(inherit.rbac().is_some());
assert_eq!(inherit.role_rules().exempt_roles, vec!["a".to_string()]);
assert_eq!(inherit.role_rules().deny_write_roles, vec!["b".to_string()]);
assert_eq!(
inherit.role_rules().unconfigured,
UnconfiguredPolicy::Closed
);
assert!(inherit.meta_policy().closed);
assert_eq!(inherit.meta_policy().roles, vec!["m".to_string()]);
assert_eq!(inherit.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
let full = base.with_policy(&overrides_all());
assert!(!full.require_context());
assert_eq!(full.profile(), Profile::Standard);
assert!(full.rbac().is_none(), "rbac:null 应清除");
assert_eq!(full.role_rules().exempt_roles, vec!["x".to_string()]);
assert_eq!(full.role_rules().deny_write_roles, vec!["y".to_string()]);
assert_eq!(full.role_rules().unconfigured, UnconfiguredPolicy::Open);
assert!(!full.meta_policy().closed);
assert_eq!(full.write_link_policy(), WriteLinkPolicy::Reject);
assert!(base.require_context());
assert_eq!(base.profile(), Profile::Text2Query);
assert!(base.rbac().is_some());
assert!(base.meta_policy().closed);
}
#[test]
fn view_register_is_readonly_err() {
let base = Registry::new();
let mut view = base.with_policy(&PolicyOverrides::default());
let err = view
.register(&json!({ "name": "A", "fields": {} }))
.expect_err("视图注册应 Err");
assert!(
err.starts_with(ERR_POLICY_VIEW_READONLY),
"前缀可匹配: {err}"
);
let err2 = view
.register_with_ctx(&json!({ "name": "A", "fields": {} }), None)
.expect_err("视图注册应 Err");
assert!(
err2.starts_with(ERR_POLICY_VIEW_READONLY),
"前缀可匹配: {err2}"
);
let err3 = view
.register_batch(
&[(json!({ "name": "A", "fields": {} }), Location::default())],
None,
)
.expect_err("视图批量注册应 Err");
assert!(
err3.starts_with(ERR_POLICY_VIEW_READONLY),
"前缀可匹配: {err3}"
);
}
#[test]
#[should_panic(expected = "ERR_POLICY_VIEW_READONLY")]
fn view_clear_panics_readonly() {
let base = Registry::new();
let mut view = base.with_policy(&PolicyOverrides::default());
view.clear();
}
#[test]
fn view_sees_catalog_snapshot() {
let mut base = Registry::new();
base.register(&json!({ "name": "A", "fields": { "x": { "type": "number" } } }))
.expect("base 注册成功");
let view = base.with_policy(&PolicyOverrides::default());
assert!(view.has("A"), "派生视图应见 base 已注册目录");
assert_eq!(view.get("A").expect("视图可取").name, "A");
assert_eq!(view.list(), base.list());
let mut base2 = Registry::new();
base2
.register(&json!({ "name": "A", "fields": {} }))
.unwrap();
let snap = base2.with_policy(&PolicyOverrides::default());
base2
.register(&json!({ "name": "B", "fields": {} }))
.unwrap();
assert!(snap.has("A"), "在途视图仍持其快照");
let fresh = base2.with_policy(&PolicyOverrides::default());
assert!(fresh.has("B"), "新派生视图应见 base 后续注册");
}
#[test]
fn view_policy_set_isolated() {
let base = Registry::new();
let mut view = base.with_policy(&PolicyOverrides::default());
view.set_profile(Profile::Text2Query);
assert_eq!(view.profile(), Profile::Text2Query);
assert_eq!(base.profile(), Profile::Standard, "视图 set 不影响 base");
let sibling = base.with_policy(&PolicyOverrides::default());
assert_eq!(sibling.profile(), Profile::Standard, "不影响兄弟视图");
}
#[test]
fn from_value_unknown_key_and_rbac_null() {
let e = PolicyOverrides::from_value(&json!({ "nope": 1 })).expect_err("未知键应 Err");
assert!(e.to_string().starts_with(ERR_POLICY_VIEW_READONLY), "{e}");
assert!(
PolicyOverrides::from_value(&json!({ "profile": "bogus" })).is_err(),
"非法 profile 应 Err(禁静默回落)"
);
assert!(
PolicyOverrides::from_value(&json!({ "writeLinkPolicy": "bogus" })).is_err(),
"非法 writeLinkPolicy 应 Err"
);
assert!(
PolicyOverrides::from_value(&json!({})).is_ok(),
"空对象合法"
);
let mut base = Registry::new();
base.set_rbac(Some(&json!({}))).unwrap();
let cleared =
base.with_policy(&PolicyOverrides::from_value(&json!({ "rbac": null })).unwrap());
assert!(cleared.rbac().is_none(), "rbac:null 应清除");
let kept = base.with_policy(&PolicyOverrides::from_value(&json!({})).unwrap());
assert!(kept.rbac().is_some(), "缺键应继承");
assert!(base.rbac().is_some(), "base 不受影响");
}
}