core_api/
ingest.rs

1use crate::db::GraphDb;
2use core_rules::{default_max_edges, Predicate, RuleDef};
3use core_storage::fs::Fs;
4use core_storage::{GraphError, Result, Value};
5use serde::Serialize;
6use std::collections::{BTreeMap, BTreeSet};
7
8/// Options for [`GraphDb::ingest`].
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct IngestOptions {
11    /// Property used as the node key. Also stored as a normal property.
12    pub key_field: String,
13    pub auto_fk: AutoFk,
14}
15
16impl Default for IngestOptions {
17    fn default() -> Self {
18        Self {
19            key_field: "id".into(),
20            auto_fk: AutoFk::default(),
21        }
22    }
23}
24
25/// Zero-config FK inference: declare a `KeyMatch` rule per `*_id` field, or skip.
26///
27/// Auto-declared rule names are `auto_fk_<src_label_lowercase>_<field>`
28/// (e.g. `auto_fk_person_org_id`, `auto_fk_device_org_id`) so two ingested
29/// labels sharing an FK field each get their own rule. A name collision is
30/// only the same `(label, field)` pair, where silent skip is correct.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub enum AutoFk {
33    Auto { suffix: String },
34    Off,
35}
36
37impl Default for AutoFk {
38    fn default() -> Self {
39        AutoFk::Auto {
40            suffix: "_id".into(),
41        }
42    }
43}
44
45/// One auto-FK field that was not turned into a rule.
46#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
47pub struct FkSkip {
48    pub field: String,
49    pub reason: String,
50}
51
52/// Outcome of one [`GraphDb::ingest`] call. Row-level issues are collected here;
53/// a commit-level `Err` means nothing was applied.
54#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
55pub struct IngestReport {
56    pub inserted: usize,
57    pub row_errors: Vec<(usize, String)>,
58    pub rules_created: Vec<String>,
59    pub skipped_fk_fields: Vec<FkSkip>,
60    /// User edges from the same request that were newly inserted (duplicates
61    /// are no-ops and do not count).
62    pub edges_inserted: usize,
63}
64
65/// Convert a JSON value to a stored [`Value`].
66///
67/// JSON `null` returns `None` so the caller can skip the field (not an error).
68/// Integral numbers become [`Value::Int`]; other numbers become [`Value::Float`].
69/// Arrays become [`Value::List`] by recursing through this function. If any
70/// element is JSON `null` the entire array field returns `None` and the caller
71/// silently skips the field (same null-drop policy as top-level fields). JSON
72/// objects become [`Value::Map`] recursively; `null` values inside an object
73/// are silently omitted from the map.
74///
75/// **Behavior change from pre-Map:** JSON objects previously returned `None`
76/// (treated as a skipped/error field by `ingest_json`). They now produce
77/// `Value::Map` so nested objects are stored faithfully.
78pub fn json_to_value(v: serde_json::Value) -> Option<Value> {
79    match v {
80        serde_json::Value::Null => None,
81        serde_json::Value::Bool(b) => Some(Value::Bool(b)),
82        serde_json::Value::Number(n) => number_to_value(&n),
83        serde_json::Value::String(s) => Some(Value::Str(s)),
84        serde_json::Value::Array(items) => {
85            let mut out = Vec::with_capacity(items.len());
86            for item in items {
87                out.push(json_to_value(item)?);
88            }
89            Some(Value::List(out))
90        }
91        serde_json::Value::Object(obj) => {
92            let mut map = std::collections::BTreeMap::new();
93            for (k, v) in obj {
94                if let Some(val) = json_to_value(v) {
95                    map.insert(k, val);
96                }
97            }
98            Some(Value::Map(map))
99        }
100    }
101}
102
103fn number_to_value(n: &serde_json::Number) -> Option<Value> {
104    if let Some(i) = n.as_i64() {
105        return Some(Value::Int(i));
106    }
107    let f = n.as_f64()?;
108    if f.is_finite() && f.fract() == 0.0 && f >= i64::MIN as f64 && f <= i64::MAX as f64 {
109        Some(Value::Int(f as i64))
110    } else {
111        Some(Value::Float(f))
112    }
113}
114
115fn object_to_row(
116    obj: &serde_json::Map<String, serde_json::Value>,
117) -> std::result::Result<BTreeMap<String, Value>, String> {
118    let mut row = BTreeMap::new();
119    for (k, v) in obj {
120        if let Some(val) = json_to_value(v.clone()) {
121            row.insert(k.clone(), val);
122        }
123    }
124    Ok(row)
125}
126
127/// Parsed JSON rows ready for [`crate::GraphDb::ingest`], plus bookkeeping so
128/// per-row shape errors keep their original JSON-array indices.
129pub struct JsonRows {
130    /// Rows that passed shape checks, in original order.
131    pub rows: Vec<BTreeMap<String, Value>>,
132    kept_indices: Vec<usize>,
133    shape_errors: Vec<(usize, String)>,
134}
135
136impl JsonRows {
137    /// Remap ingest row-error indices onto the original JSON array and append
138    /// the shape errors collected by [`json_to_rows`].
139    pub fn into_report(self, mut report: IngestReport) -> IngestReport {
140        for (idx, _) in &mut report.row_errors {
141            *idx = self.kept_indices[*idx];
142        }
143        report.row_errors.extend(self.shape_errors);
144        report.row_errors.sort_by_key(|(i, _)| *i);
145        report
146    }
147}
148
149/// Convert a parsed JSON value (must be an array of objects) into ingest rows.
150///
151/// Same conversion as [`crate::GraphDb::ingest_json`]: [`json_to_value`] per
152/// field; nested objects / mixed arrays become per-row errors. A top-level
153/// value that is not an array of objects is [`GraphError::IngestError`].
154pub fn json_to_rows(value: &serde_json::Value) -> Result<JsonRows> {
155    let arr = value.as_array().ok_or_else(|| GraphError::IngestError {
156        detail: "top-level JSON must be an array of objects".into(),
157    })?;
158    if !arr.iter().all(|v| v.is_object()) {
159        return Err(GraphError::IngestError {
160            detail: "top-level JSON must be an array of objects".into(),
161        });
162    }
163
164    let mut rows = Vec::new();
165    let mut shape_errors = Vec::new();
166    let mut kept_indices = Vec::new();
167    for (i, item) in arr.iter().enumerate() {
168        let obj = item
169            .as_object()
170            .expect("top-level checked as array of objects");
171        match object_to_row(obj) {
172            Ok(row) => {
173                kept_indices.push(i);
174                rows.push(row);
175            }
176            Err(msg) => shape_errors.push((i, msg)),
177        }
178    }
179    Ok(JsonRows {
180        rows,
181        kept_indices,
182        shape_errors,
183    })
184}
185
186/// Parse JSON, convert rows, then delegate to [`run`].
187pub(crate) fn run_json<F: Fs>(
188    db: &mut GraphDb<F>,
189    label: &str,
190    json: &str,
191    opts: &IngestOptions,
192) -> Result<IngestReport> {
193    let parsed: serde_json::Value =
194        serde_json::from_str(json).map_err(|e| GraphError::IngestError {
195            detail: e.to_string(),
196        })?;
197    let mut converted = json_to_rows(&parsed)?;
198    let rows = std::mem::take(&mut converted.rows);
199    let report = run(db, label, rows, opts, &[])?;
200    Ok(converted.into_report(report))
201}
202
203type PropMap = BTreeMap<String, Value>;
204
205struct Classified {
206    accepted: Vec<(String, PropMap)>,
207    row_errors: Vec<(usize, String)>,
208}
209
210/// Classify rows, optionally infer auto-FK rules, and commit one atomic batch
211/// (rules first, then node inserts, then optional user edges).
212pub(crate) fn run<F: Fs>(
213    db: &mut GraphDb<F>,
214    label: &str,
215    rows: Vec<BTreeMap<String, Value>>,
216    opts: &IngestOptions,
217    edges: &[(String, String, String)],
218) -> Result<IngestReport> {
219    let Classified {
220        accepted,
221        row_errors,
222    } = classify_rows(db, rows, &opts.key_field);
223
224    let (new_rules, skipped_fk_fields) = match &opts.auto_fk {
225        AutoFk::Off => (Vec::new(), Vec::new()),
226        AutoFk::Auto { suffix } => infer_auto_fk(db, label, suffix, &opts.key_field, &accepted),
227    };
228
229    let rules_created: Vec<String> = new_rules.iter().map(|r| r.name.clone()).collect();
230
231    // One `WalRecord::Batch` (Batched fsync), not a loop of `insert_node`.
232    let mut batch = db.batch();
233    for def in new_rules {
234        batch.create_rule(def);
235    }
236    for (key, props) in &accepted {
237        let prop_vec: Vec<(String, Value)> =
238            props.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
239        batch.insert_node(label, key, prop_vec);
240    }
241    for (etype, src, dst) in edges {
242        batch.insert_edge(etype, src, dst);
243    }
244    let (_, edges_inserted) = batch.commit_ingest(label, accepted.len())?;
245
246    Ok(IngestReport {
247        inserted: accepted.len(),
248        row_errors,
249        rules_created,
250        skipped_fk_fields,
251        edges_inserted,
252    })
253}
254
255fn classify_rows<F: Fs>(db: &GraphDb<F>, rows: Vec<PropMap>, key_field: &str) -> Classified {
256    let mut accepted = Vec::new();
257    let mut row_errors = Vec::new();
258    let mut seen: BTreeSet<String> = BTreeSet::new();
259
260    for (i, row) in rows.into_iter().enumerate() {
261        match row.get(key_field) {
262            None => row_errors.push((i, format!("missing key field {key_field}"))),
263            Some(Value::Str(key)) => {
264                if db.has_node(key) || seen.contains(key) {
265                    row_errors.push((i, format!("duplicate key {key}")));
266                } else {
267                    seen.insert(key.clone());
268                    accepted.push((key.clone(), row));
269                }
270            }
271            Some(_) => row_errors.push((i, format!("key field {key_field} is not a string"))),
272        }
273    }
274    Classified {
275        accepted,
276        row_errors,
277    }
278}
279
280fn infer_auto_fk<F: Fs>(
281    db: &GraphDb<F>,
282    src_label: &str,
283    suffix: &str,
284    key_field: &str,
285    accepted: &[(String, PropMap)],
286) -> (Vec<RuleDef>, Vec<FkSkip>) {
287    let existing_rule_names: BTreeSet<String> = db.rules().into_iter().map(|r| r.name).collect();
288    let accepted_keys: BTreeSet<&str> = accepted.iter().map(|(k, _)| k.as_str()).collect();
289
290    let mut fields: BTreeSet<String> = BTreeSet::new();
291    for (_, row) in accepted {
292        for field in row.keys() {
293            if field != key_field && field.ends_with(suffix) && field.len() > suffix.len() {
294                fields.insert(field.clone());
295            }
296        }
297    }
298
299    let mut new_rules = Vec::new();
300    let mut skipped = Vec::new();
301
302    for field in fields {
303        let mut values: BTreeSet<&str> = BTreeSet::new();
304        for (_, row) in accepted {
305            if let Some(Value::Str(s)) = row.get(&field) {
306                values.insert(s.as_str());
307            }
308        }
309
310        let mut labels: BTreeSet<String> = BTreeSet::new();
311        for value in values {
312            if let Some(n) = db.node_ref(value) {
313                labels.insert(n.label().to_string());
314            }
315            if accepted_keys.contains(value) {
316                labels.insert(src_label.to_string());
317            }
318        }
319
320        match labels.len() {
321            0 => skipped.push(FkSkip {
322                field,
323                reason: "no matching target keys".into(),
324            }),
325            1 => {
326                let dst_label = labels.into_iter().next().expect("len == 1");
327                // `auto_fk_<src_label_lowercase>_<field>` — scoped by source
328                // label so Person.org_id and Device.org_id do not collide.
329                let name = format!("auto_fk_{}_{field}", src_label.to_lowercase());
330                if existing_rule_names.contains(&name) {
331                    continue;
332                }
333                let remainder = &field[..field.len() - suffix.len()];
334                let predicate = Predicate::KeyMatch {
335                    field: field.clone(),
336                };
337                let max_edges = Some(default_max_edges(&predicate));
338                new_rules.push(RuleDef {
339                    name,
340                    src_label: src_label.to_string(),
341                    dst_label,
342                    predicate,
343                    edge_type: remainder.to_uppercase(),
344                    weight_prop: None,
345                    max_edges,
346                    approximate: false,
347                    via_label: None,
348                    via_edge: None,
349                    via_dir: None,
350                    namespace: None,
351                });
352            }
353            _ => {
354                let listed = labels.into_iter().collect::<Vec<_>>().join(", ");
355                skipped.push(FkSkip {
356                    field,
357                    reason: format!("ambiguous target labels: {listed}"),
358                });
359            }
360        }
361    }
362
363    (new_rules, skipped)
364}
core_api/ingest.rs

core_api/
ingest.rs
