1use serde_json::Value;
11
12use crate::datasource::DataSource;
13use crate::dialect::Backend;
14
15use super::definition::{Location, Role};
16use super::registry::{classify_batch, detect_location_conflict, BatchError};
17
18#[derive(Debug, Clone)]
20pub struct SourceDecl {
21 pub name: String,
22 pub kind: String,
24 pub databases: Vec<String>,
26}
27
28#[derive(Debug, Clone)]
33pub struct LinkDecl {
34 pub source: String,
36 pub role: Option<Role>,
38 pub priority: Option<i64>,
40 pub database: Option<String>,
42 pub schema: Option<String>,
44}
45
46#[derive(Debug, Clone, Default)]
48pub struct LoadConfig {
49 pub sources: Vec<SourceDecl>,
51 pub defs: Vec<String>,
52 pub links: Vec<(String, Vec<LinkDecl>)>,
54}
55
56impl LoadConfig {
57 pub fn from_json(v: &Value) -> Result<Self, String> {
63 let obj = v
64 .as_object()
65 .ok_or_else(|| "ERR:LOAD store.config.json 必须是对象".to_string())?;
66
67 let sources_obj = obj
68 .get("sources")
69 .and_then(|s| s.as_object())
70 .ok_or_else(|| {
71 "ERR:LOAD 缺 sources(须为 { name: { kind, databases } } 对象)".to_string()
72 })?;
73 if sources_obj.is_empty() {
74 return Err("ERR:LOAD sources 不得为空".to_string());
75 }
76
77 let mut sources = Vec::new();
78 for (name, decl) in sources_obj {
79 let d = decl
80 .as_object()
81 .ok_or_else(|| format!("ERR:LOAD 连接 {name} 声明必须是对象"))?;
82 let raw_kind = d
83 .get("kind")
84 .and_then(|k| k.as_str())
85 .filter(|s| !s.is_empty())
86 .ok_or_else(|| format!("ERR:LOAD 连接 {name} 缺 kind"))?;
87 DataSource::from_kind(raw_kind).map_err(|_| {
89 format!(
90 "ERR:LOAD 连接 {name} 的 kind 非法: {raw_kind}(仅 mongodb/mysql/pg/sqlite/local/memory)"
91 )
92 })?;
93 let databases: Vec<String> = d
94 .get("databases")
95 .and_then(|ds| ds.as_array())
96 .map(|arr| {
97 arr.iter()
98 .filter_map(|x| x.as_str())
99 .filter(|s| !s.is_empty())
100 .map(String::from)
101 .collect()
102 })
103 .unwrap_or_default();
104 if databases.is_empty() {
105 return Err(format!(
106 "ERR:LOAD 连接 {name} 的 databases 须为非空字符串数组"
107 ));
108 }
109 sources.push(SourceDecl {
110 name: name.clone(),
111 kind: raw_kind.to_lowercase(),
112 databases,
113 });
114 }
115
116 let defs: Vec<String> = obj
117 .get("defs")
118 .and_then(|ds| ds.as_array())
119 .map(|arr| {
120 arr.iter()
121 .filter_map(|x| x.as_str())
122 .filter(|s| !s.is_empty())
123 .map(String::from)
124 .collect()
125 })
126 .unwrap_or_default();
127 if defs.is_empty() {
128 return Err("ERR:LOAD defs 须为非空字符串数组(定义根)".to_string());
129 }
130
131 let links = match obj.get("links") {
133 None | Some(Value::Null) => Vec::new(),
134 Some(v) => {
135 let links_obj = v.as_object().ok_or_else(|| {
136 "ERR:LOAD links 必须是对象(schema 名 → 链路声明数组)".to_string()
137 })?;
138 let mut out: Vec<(String, Vec<LinkDecl>)> = Vec::with_capacity(links_obj.len());
139 for (name, decls_v) in links_obj {
140 let arr = decls_v.as_array().ok_or_else(|| {
141 format!("ERR:LOAD links[{name}] 必须是数组(链路声明列表)")
142 })?;
143 let mut decls: Vec<LinkDecl> = Vec::with_capacity(arr.len());
144 let mut primary_count = 0usize;
145 for (i, item) in arr.iter().enumerate() {
146 let d = item
147 .as_object()
148 .ok_or_else(|| format!("ERR:LOAD links[{name}][{i}] 必须是对象"))?;
149 let source = d
151 .get("source")
152 .and_then(|s| s.as_str())
153 .filter(|s| !s.is_empty())
154 .ok_or_else(|| format!("ERR:LOAD links[{name}][{i}] 缺 source"))?;
155 let src_decl = sources.iter().find(|s| s.name == source).ok_or_else(|| {
156 format!("ERR:LOAD links[{name}] 的 source 未在 sources 中声明: {source}")
157 })?;
158 let role = match d.get("role") {
160 None | Some(Value::Null) => None,
161 Some(rv) => {
162 let rs = rv.as_str().ok_or_else(|| {
163 format!("ERR:LOAD links[{name}][{i}] 的 role 须为字符串")
164 })?;
165 Some(Role::parse(rs)?)
166 }
167 };
168 if role == Some(Role::Primary) {
169 primary_count += 1;
170 }
171 let priority = match d.get("priority") {
173 None | Some(Value::Null) => None,
174 Some(pv) => Some(pv.as_i64().ok_or_else(|| {
175 format!("ERR:LOAD links[{name}][{i}] 的 priority 须为整数")
176 })?),
177 };
178 let database = match d.get("database") {
180 None | Some(Value::Null) => {
181 if src_decl.databases.len() == 1 {
182 Some(src_decl.databases[0].clone())
183 } else {
184 return Err(format!(
185 "ERR:LOAD links[{name}] 的 source {source} 维护多库,须显式给 database"
186 ));
187 }
188 }
189 Some(dv) => Some(
190 dv.as_str()
191 .filter(|s| !s.is_empty())
192 .ok_or_else(|| {
193 format!(
194 "ERR:LOAD links[{name}][{i}] 的 database 须为非空字符串"
195 )
196 })?
197 .to_string(),
198 ),
199 };
200 let schema = match d.get("schema") {
201 None | Some(Value::Null) => None,
202 Some(sv) => Some(
203 sv.as_str()
204 .filter(|s| !s.is_empty())
205 .ok_or_else(|| {
206 format!(
207 "ERR:LOAD links[{name}][{i}] 的 schema 须为非空字符串"
208 )
209 })?
210 .to_string(),
211 ),
212 };
213 decls.push(LinkDecl {
214 source: source.to_string(),
215 role,
216 priority,
217 database,
218 schema,
219 });
220 }
221 if primary_count != 1 {
223 return Err(format!(
224 "ERR:LOAD links[{name}] 必须恰好一条 role=primary(实际 {primary_count} 条)"
225 ));
226 }
227 out.push((name.clone(), decls));
228 }
229 out
230 }
231 };
232
233 Ok(LoadConfig {
234 sources,
235 defs,
236 links,
237 })
238 }
239
240 pub fn source_of_db(&self, db: &str) -> Result<(&SourceDecl, u8), String> {
244 let mut hit: Option<&SourceDecl> = None;
245 for s in &self.sources {
246 if s.databases.iter().any(|d| d == db) {
247 if let Some(prev) = hit {
248 return Err(format!(
249 "ERR:LOAD 库归属冲突: {db} 同时声明在连接 {} 与 {}(同一 db 名只能挂一个连接)",
250 prev.name, s.name
251 ));
252 }
253 hit = Some(s);
254 }
255 }
256 let decl = hit.ok_or_else(|| {
257 format!(
258 "ERR:LOAD 库目录未声明: {db}(须在 store.config.json 的某连接 databases 中列出)"
259 )
260 })?;
261 let depth = depth_layered(&decl.kind)?;
262 Ok((decl, depth))
263 }
264}
265
266pub fn depth_layered(kind: &str) -> Result<u8, String> {
269 let ds = DataSource::from_kind(kind).map_err(|_| {
270 format!("ERR:LOAD 连接 kind 非法: {kind}(仅 mongodb/mysql/pg/sqlite/local/memory)")
271 })?;
272 Ok(match ds.backend() {
273 Some(Backend::Postgres) => 2,
274 _ => 1,
275 })
276}
277
278pub fn locate(rel_under_db: &str, source: &str, db: &str, depth: u8) -> Result<Location, String> {
285 let segs: Vec<&str> = rel_under_db
286 .split(['/', '\\'])
287 .filter(|s| !s.is_empty())
288 .collect();
289 if segs.is_empty() {
290 return Err(format!("ERR:LOAD 定义路径非法(缺文件名): {rel_under_db}"));
291 }
292 let dirs = &segs[..segs.len() - 1];
294 let schema = if depth >= 2 && !dirs.is_empty() {
295 Some(dirs[0].to_string())
296 } else {
297 None
298 };
299 Ok(Location {
300 source: source.to_string(),
301 database: Some(db.to_string()),
302 schema,
303 role: None,
304 priority: None,
305 })
306}
307
308pub fn plan_load(
317 cfg: &LoadConfig,
318 files: &[(String, Value)],
319) -> Result<Vec<(Value, Location)>, String> {
320 let mut items: Vec<(Value, Location)> = Vec::with_capacity(files.len());
322 for (rel, defn) in files {
323 let segs: Vec<&str> = rel.split(['/', '\\']).filter(|s| !s.is_empty()).collect();
324 if segs.is_empty() {
325 return Err(format!("ERR:LOAD 定义路径非法: {rel}"));
326 }
327 let db = segs[0];
328 let (decl, depth) = cfg.source_of_db(db)?;
329 let rel_under_db = segs[1..].join("/");
330 let loc = locate(&rel_under_db, &decl.name, db, depth)?;
331 items.push((defn.clone(), loc));
332 }
333
334 let class = classify_batch(&items).map_err(|e| batch_err_to_load(e, files))?;
336
337 let mut names: Vec<&String> = class.order.iter().collect();
339 names.sort();
340
341 let mut quads: Vec<(String, String, Location)> = Vec::new();
343 for name in &names {
344 if let Some((_, loc, coll)) = class.primaries.get(*name) {
345 quads.push(((*name).clone(), coll.clone(), loc.clone()));
346 }
347 if let Some(rs) = class.replicas.get(*name) {
348 for (defn, loc) in rs {
349 let coll = defn
350 .get("name")
351 .and_then(|v| v.as_str())
352 .unwrap_or(name)
353 .to_string();
354 quads.push(((*name).clone(), coll, loc.clone()));
355 }
356 }
357 }
358 if let Some(c) = detect_location_conflict(&quads) {
359 return Err(format!(
360 "ERR:LOAD 落点冲突: {} 同时映射 ({}, {:?}, {:?})(同一落点只能有一个 collection;冲突名 `{}` 与 `{}`)",
361 c.collection, c.source, c.database, c.schema, c.prev, c.name
362 ));
363 }
364
365 let mut out: Vec<(Value, Location)> = Vec::with_capacity(items.len());
366 for name in names {
367 let Some((defn, loc, _coll)) = class.primaries.get(name) else {
368 return Err(format!(
369 "ERR:LOAD 主定义缺失: {name}(同名项均为 replica:true,须恰好一份主)"
370 ));
371 };
372 out.push((defn.clone(), loc.clone()));
373 if let Some(rs) = class.replicas.get(name) {
374 for (rdefn, rloc) in rs {
375 out.push((rdefn.clone(), rloc.clone()));
376 }
377 }
378 }
379 Ok(out)
380}
381
382fn batch_err_to_load(e: BatchError, files: &[(String, Value)]) -> String {
384 match e {
385 BatchError::Parse(m) => format!("ERR:LOAD {m}"),
386 BatchError::PrimaryDuplicate { name, indices } => {
387 let rels: Vec<String> = indices
388 .iter()
389 .map(|i| {
390 files
391 .get(*i)
392 .map(|(r, _)| r.clone())
393 .unwrap_or_else(|| format!("#{i}"))
394 })
395 .collect();
396 format!(
397 "ERR:LOAD 主定义重复: {name}({})——同名只允许一份主,其余须 replica:true",
398 rels.join(" 与 ")
399 )
400 }
401 BatchError::LocationConflict(c) => format!(
402 "ERR:LOAD 落点冲突: {} 同时映射 ({}, {:?}, {:?})(同一落点只能有一个 collection;冲突名 `{}` 与 `{}`)",
403 c.collection, c.source, c.database, c.schema, c.prev, c.name
404 ),
405 }
406}