use std::collections::HashMap;
use serde_json::{json, Map, Value};
use crate::types::{is_truthy, str_list, AGG_OPS};
use super::definition::{normalize_fields, ComputeDef, FieldDef, RelationDef, Schema};
#[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, Default)]
pub struct Registry {
schemas: HashMap<String, Schema>,
order: Vec<String>,
require_context: bool,
profile: Profile,
}
impl Registry {
pub fn new() -> Self {
Self::default()
}
pub fn register(&mut self, defn: &Value) -> Result<(), String> {
let obj = defn
.as_object()
.ok_or_else(|| "schema 定义必须是对象".to_string())?;
let name = obj
.get("name")
.and_then(|v| v.as_str())
.ok_or_else(|| "schema 缺少 name".to_string())?
.to_string();
let collection = obj
.get("collection")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.map(String::from)
.unwrap_or_else(|| name.clone());
let schema = build_schema(obj, name.clone(), collection.clone())?;
self.check_location_unique(&name, &schema)?;
let is_archive = obj.get("_isArchive").map(is_truthy).unwrap_or(false);
self.schemas.insert(name.clone(), schema);
self.order.push(name.clone());
if !is_archive && !name.ends_with("Deleted") {
self.register(&archive_defn(obj, &name, &collection))?;
}
Ok(())
}
pub fn get(&self, name: &str) -> Result<&Schema, String> {
self.schemas
.get(name)
.ok_or_else(|| format!("Schema 未注册: {}", name))
}
pub fn has(&self, name: &str) -> bool {
self.schemas.contains_key(name)
}
pub fn set_require_context(&mut self, require: bool) {
self.require_context = require;
}
pub fn require_context(&self) -> bool {
self.require_context
}
pub fn set_profile(&mut self, profile: Profile) {
self.profile = profile;
}
pub fn profile(&self) -> Profile {
self.profile
}
pub fn get_by_location(
&self,
source: &str,
namespace: Option<&str>,
collection: &str,
) -> Result<&Schema, String> {
let ns = namespace.filter(|s| !s.is_empty());
self.schemas
.values()
.find(|s| s.collection == collection && s.source() == source && s.ns() == ns)
.ok_or_else(|| {
format!(
"Schema 未定位(source = {}, namespace = {:?}, collection = {})",
source, ns, collection
)
})
}
pub fn get_for_command(
&self,
source: &str,
namespace: Option<&str>,
collection: &str,
) -> Result<&Schema, String> {
if let Ok(s) = self.get_by_location(source, namespace, collection) {
return Ok(s);
}
let candidates: Vec<&Schema> = self
.schemas
.values()
.filter(|s| s.collection == collection && s.source() == source)
.collect();
match candidates.len() {
0 => self.get_by_location(source, namespace, collection),
1 => Ok(candidates[0]),
_ => candidates
.iter()
.find(|s| s.ns().is_none())
.copied()
.ok_or_else(|| {
format!(
"命令定位 ({}, {:?}, {}) 存在多个结构 schema(多租户 override 需唯一的 (source, collection) 结构声明)",
source, namespace, collection
)
}),
}
}
fn check_location_unique(&self, name: &str, schema: &Schema) -> Result<(), String> {
let conflict = self.schemas.iter().find(|(n, s)| {
n.as_str() != name
&& s.collection == schema.collection
&& s.source() == schema.source()
&& s.ns() == schema.ns()
});
match conflict {
None => Ok(()),
Some((other, s)) => Err(format!(
"定位三元组冲突: ({}, {:?}, {}) 已被 schema `{}`(collection = {})占用",
schema.source(),
schema.ns(),
schema.collection,
other,
s.collection
)),
}
}
pub fn list(&self) -> Vec<String> {
self.order.clone()
}
pub fn schema_datasource(&self, name: &str) -> Result<Option<String>, String> {
Ok(self.get(name)?.datasource.clone())
}
pub fn resolve_datasource(
&self,
schema_name: &str,
config: &Value,
) -> Result<crate::datasource::DataSource, String> {
let declared = self.get(schema_name)?.datasource.as_deref();
let cfg = crate::datasource::DataSourceConfig::from_json(config)?;
cfg.resolve(declared)
}
}
fn build_schema(
obj: &Map<String, Value>,
name: String,
collection: String,
) -> Result<Schema, String> {
let timestamps = parse_timestamps(obj)?;
let mut fields = normalize_fields(obj.get("fields"))?;
if timestamps {
add_timestamp_fields(&mut fields);
}
let relations = parse_relations(obj);
let computes = parse_computes(obj, &relations)?;
Ok(Schema {
name,
collection,
id_prefix: obj
.get("idPrefix")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string(),
timestamps,
fields,
relations,
computes,
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(),
datasource: opt_nonempty_str(obj.get("datasource")),
namespace: opt_nonempty_str(obj.get("namespace")),
})
}
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,
});
}
}
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 opt_nonempty_str(v: Option<&Value>) -> Option<String> {
v.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.map(String::from)
}
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" }));
let mut 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!([])),
});
if let Some(ds) = obj.get("datasource") {
arch["datasource"] = ds.clone();
}
if let Some(ns) = obj.get("namespace") {
arch["namespace"] = ns.clone();
}
arch
}