use serde_json::Value;
use crate::datasource::DataSource;
use crate::dialect::Backend;
use super::definition::{Location, Role};
use super::registry::{classify_batch, detect_location_conflict, BatchError};
#[derive(Debug, Clone)]
pub struct SourceDecl {
pub name: String,
pub kind: String,
pub databases: Vec<String>,
}
#[derive(Debug, Clone)]
pub struct LinkDecl {
pub source: String,
pub role: Option<Role>,
pub priority: Option<i64>,
pub database: Option<String>,
pub schema: Option<String>,
}
#[derive(Debug, Clone, Default)]
pub struct LoadConfig {
pub sources: Vec<SourceDecl>,
pub defs: Vec<String>,
pub links: Vec<(String, Vec<LinkDecl>)>,
}
impl LoadConfig {
pub fn from_json(v: &Value) -> Result<Self, String> {
let obj = v
.as_object()
.ok_or_else(|| "ERR:LOAD store.config.json 必须是对象".to_string())?;
let sources_obj = obj
.get("sources")
.and_then(|s| s.as_object())
.ok_or_else(|| {
"ERR:LOAD 缺 sources(须为 { name: { kind, databases } } 对象)".to_string()
})?;
if sources_obj.is_empty() {
return Err("ERR:LOAD sources 不得为空".to_string());
}
let mut sources = Vec::new();
for (name, decl) in sources_obj {
let d = decl
.as_object()
.ok_or_else(|| format!("ERR:LOAD 连接 {name} 声明必须是对象"))?;
let raw_kind = d
.get("kind")
.and_then(|k| k.as_str())
.filter(|s| !s.is_empty())
.ok_or_else(|| format!("ERR:LOAD 连接 {name} 缺 kind"))?;
DataSource::from_kind(raw_kind).map_err(|_| {
format!(
"ERR:LOAD 连接 {name} 的 kind 非法: {raw_kind}(仅 mongodb/mysql/pg/sqlite/local/memory)"
)
})?;
let databases: Vec<String> = d
.get("databases")
.and_then(|ds| ds.as_array())
.map(|arr| {
arr.iter()
.filter_map(|x| x.as_str())
.filter(|s| !s.is_empty())
.map(String::from)
.collect()
})
.unwrap_or_default();
if databases.is_empty() {
return Err(format!(
"ERR:LOAD 连接 {name} 的 databases 须为非空字符串数组"
));
}
sources.push(SourceDecl {
name: name.clone(),
kind: raw_kind.to_lowercase(),
databases,
});
}
let defs: Vec<String> = obj
.get("defs")
.and_then(|ds| ds.as_array())
.map(|arr| {
arr.iter()
.filter_map(|x| x.as_str())
.filter(|s| !s.is_empty())
.map(String::from)
.collect()
})
.unwrap_or_default();
if defs.is_empty() {
return Err("ERR:LOAD defs 须为非空字符串数组(定义根)".to_string());
}
let links = match obj.get("links") {
None | Some(Value::Null) => Vec::new(),
Some(v) => {
let links_obj = v.as_object().ok_or_else(|| {
"ERR:LOAD links 必须是对象(schema 名 → 链路声明数组)".to_string()
})?;
let mut out: Vec<(String, Vec<LinkDecl>)> = Vec::with_capacity(links_obj.len());
for (name, decls_v) in links_obj {
let arr = decls_v.as_array().ok_or_else(|| {
format!("ERR:LOAD links[{name}] 必须是数组(链路声明列表)")
})?;
let mut decls: Vec<LinkDecl> = Vec::with_capacity(arr.len());
let mut primary_count = 0usize;
for (i, item) in arr.iter().enumerate() {
let d = item
.as_object()
.ok_or_else(|| format!("ERR:LOAD links[{name}][{i}] 必须是对象"))?;
let source = d
.get("source")
.and_then(|s| s.as_str())
.filter(|s| !s.is_empty())
.ok_or_else(|| format!("ERR:LOAD links[{name}][{i}] 缺 source"))?;
let src_decl = sources.iter().find(|s| s.name == source).ok_or_else(|| {
format!("ERR:LOAD links[{name}] 的 source 未在 sources 中声明: {source}")
})?;
let role = match d.get("role") {
None | Some(Value::Null) => None,
Some(rv) => {
let rs = rv.as_str().ok_or_else(|| {
format!("ERR:LOAD links[{name}][{i}] 的 role 须为字符串")
})?;
Some(Role::parse(rs)?)
}
};
if role == Some(Role::Primary) {
primary_count += 1;
}
let priority = match d.get("priority") {
None | Some(Value::Null) => None,
Some(pv) => Some(pv.as_i64().ok_or_else(|| {
format!("ERR:LOAD links[{name}][{i}] 的 priority 须为整数")
})?),
};
let database = match d.get("database") {
None | Some(Value::Null) => {
if src_decl.databases.len() == 1 {
Some(src_decl.databases[0].clone())
} else {
return Err(format!(
"ERR:LOAD links[{name}] 的 source {source} 维护多库,须显式给 database"
));
}
}
Some(dv) => Some(
dv.as_str()
.filter(|s| !s.is_empty())
.ok_or_else(|| {
format!(
"ERR:LOAD links[{name}][{i}] 的 database 须为非空字符串"
)
})?
.to_string(),
),
};
let schema = match d.get("schema") {
None | Some(Value::Null) => None,
Some(sv) => Some(
sv.as_str()
.filter(|s| !s.is_empty())
.ok_or_else(|| {
format!(
"ERR:LOAD links[{name}][{i}] 的 schema 须为非空字符串"
)
})?
.to_string(),
),
};
decls.push(LinkDecl {
source: source.to_string(),
role,
priority,
database,
schema,
});
}
if primary_count != 1 {
return Err(format!(
"ERR:LOAD links[{name}] 必须恰好一条 role=primary(实际 {primary_count} 条)"
));
}
out.push((name.clone(), decls));
}
out
}
};
Ok(LoadConfig {
sources,
defs,
links,
})
}
pub fn source_of_db(&self, db: &str) -> Result<(&SourceDecl, u8), String> {
let mut hit: Option<&SourceDecl> = None;
for s in &self.sources {
if s.databases.iter().any(|d| d == db) {
if let Some(prev) = hit {
return Err(format!(
"ERR:LOAD 库归属冲突: {db} 同时声明在连接 {} 与 {}(同一 db 名只能挂一个连接)",
prev.name, s.name
));
}
hit = Some(s);
}
}
let decl = hit.ok_or_else(|| {
format!(
"ERR:LOAD 库目录未声明: {db}(须在 store.config.json 的某连接 databases 中列出)"
)
})?;
let depth = depth_layered(&decl.kind)?;
Ok((decl, depth))
}
}
pub fn depth_layered(kind: &str) -> Result<u8, String> {
let ds = DataSource::from_kind(kind).map_err(|_| {
format!("ERR:LOAD 连接 kind 非法: {kind}(仅 mongodb/mysql/pg/sqlite/local/memory)")
})?;
Ok(match ds.backend() {
Some(Backend::Postgres) => 2,
_ => 1,
})
}
pub fn locate(rel_under_db: &str, source: &str, db: &str, depth: u8) -> Result<Location, String> {
let segs: Vec<&str> = rel_under_db
.split(['/', '\\'])
.filter(|s| !s.is_empty())
.collect();
if segs.is_empty() {
return Err(format!("ERR:LOAD 定义路径非法(缺文件名): {rel_under_db}"));
}
let dirs = &segs[..segs.len() - 1];
let schema = if depth >= 2 && !dirs.is_empty() {
Some(dirs[0].to_string())
} else {
None
};
Ok(Location {
source: source.to_string(),
database: Some(db.to_string()),
schema,
role: None,
priority: None,
})
}
pub fn plan_load(
cfg: &LoadConfig,
files: &[(String, Value)],
) -> Result<Vec<(Value, Location)>, String> {
let mut items: Vec<(Value, Location)> = Vec::with_capacity(files.len());
for (rel, defn) in files {
let segs: Vec<&str> = rel.split(['/', '\\']).filter(|s| !s.is_empty()).collect();
if segs.is_empty() {
return Err(format!("ERR:LOAD 定义路径非法: {rel}"));
}
let db = segs[0];
let (decl, depth) = cfg.source_of_db(db)?;
let rel_under_db = segs[1..].join("/");
let loc = locate(&rel_under_db, &decl.name, db, depth)?;
items.push((defn.clone(), loc));
}
let class = classify_batch(&items).map_err(|e| batch_err_to_load(e, files))?;
let mut names: Vec<&String> = class.order.iter().collect();
names.sort();
let mut quads: Vec<(String, String, Location)> = Vec::new();
for name in &names {
if let Some((_, loc, coll)) = class.primaries.get(*name) {
quads.push(((*name).clone(), coll.clone(), loc.clone()));
}
if let Some(rs) = class.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(format!(
"ERR:LOAD 落点冲突: {} 同时映射 ({}, {:?}, {:?})(同一落点只能有一个 collection;冲突名 `{}` 与 `{}`)",
c.collection, c.source, c.database, c.schema, c.prev, c.name
));
}
let mut out: Vec<(Value, Location)> = Vec::with_capacity(items.len());
for name in names {
let Some((defn, loc, _coll)) = class.primaries.get(name) else {
return Err(format!(
"ERR:LOAD 主定义缺失: {name}(同名项均为 replica:true,须恰好一份主)"
));
};
out.push((defn.clone(), loc.clone()));
if let Some(rs) = class.replicas.get(name) {
for (rdefn, rloc) in rs {
out.push((rdefn.clone(), rloc.clone()));
}
}
}
Ok(out)
}
fn batch_err_to_load(e: BatchError, files: &[(String, Value)]) -> String {
match e {
BatchError::Parse(m) => format!("ERR:LOAD {m}"),
BatchError::PrimaryDuplicate { name, indices } => {
let rels: Vec<String> = indices
.iter()
.map(|i| {
files
.get(*i)
.map(|(r, _)| r.clone())
.unwrap_or_else(|| format!("#{i}"))
})
.collect();
format!(
"ERR:LOAD 主定义重复: {name}({})——同名只允许一份主,其余须 replica:true",
rels.join(" 与 ")
)
}
BatchError::LocationConflict(c) => format!(
"ERR:LOAD 落点冲突: {} 同时映射 ({}, {:?}, {:?})(同一落点只能有一个 collection;冲突名 `{}` 与 `{}`)",
c.collection, c.source, c.database, c.schema, c.prev, c.name
),
}
}