nodedb_lite/nodedb/collection/
import.rs

1//! COPY FROM: NDJSON and CSV file import into collections.
2
3use nodedb_types::error::{NodeDbError, NodeDbResult};
4
5use super::super::{LockExt, NodeDbLite};
6use crate::storage::engine::StorageEngine;
7
8impl<S: StorageEngine> NodeDbLite<S> {
9    /// Import documents from NDJSON (newline-delimited JSON) text.
10    ///
11    /// Each line is a JSON object. The "id" field is used as document ID;
12    /// if missing, a UUID v7 is generated.
13    ///
14    /// Returns the number of documents imported.
15    pub async fn copy_from_ndjson(&self, collection: &str, ndjson: &str) -> NodeDbResult<u64> {
16        let guard = self.fts_state.admit_mutation().await;
17        let result = async {
18            let mut docs: Vec<(String, Vec<(String, loro::LoroValue)>)> = Vec::new();
19            for line in ndjson.lines() {
20                let line = line.trim();
21                if line.is_empty() || line.starts_with('#') {
22                    continue;
23                }
24                let obj: serde_json::Value = sonic_rs::from_str(line)
25                    .map_err(|e| NodeDbError::bad_request(format!("invalid JSON: {e}")))?;
26
27                let id = obj
28                    .get("id")
29                    .and_then(|v| v.as_str())
30                    .map(|s| s.to_string())
31                    .unwrap_or_else(nodedb_types::id_gen::uuid_v7);
32
33                let mut fields = Vec::new();
34                if let serde_json::Value::Object(map) = &obj {
35                    for (k, v) in map {
36                        if k == "id" {
37                            continue;
38                        }
39                        fields.push((k.clone(), json_to_loro(v)));
40                    }
41                }
42                docs.push((id, fields));
43            }
44
45            let mut crdt = self.crdt.lock_or_recover();
46            let count = docs.len() as u64;
47
48            use crate::engine::crdt::engine::CrdtField;
49            let borrowed_fields: Vec<Vec<CrdtField<'_>>> = docs
50                .iter()
51                .map(|(_, fields)| {
52                    fields
53                        .iter()
54                        .map(|(k, v)| (k.as_str(), v.clone()))
55                        .collect()
56                })
57                .collect();
58
59            let ops: Vec<(&str, &str, &[CrdtField<'_>])> = docs
60                .iter()
61                .zip(borrowed_fields.iter())
62                .map(|((id, _), fields)| (collection, id.as_str(), fields.as_slice()))
63                .collect();
64
65            crdt.batch_upsert(&ops).map_err(NodeDbError::from)?;
66            drop(crdt);
67            self.reindex_documents_local(collection, docs.iter().map(|(id, _)| id.as_str()))?;
68            Ok(count)
69        }
70        .await;
71        guard.finish(result)
72    }
73
74    /// Import documents from CSV text.
75    ///
76    /// First line is the header (column names). Each subsequent line is a row.
77    /// An "id" column is used as document ID; if missing, UUIDs are generated.
78    ///
79    /// Two-pass: first pass infers column types (int/float/bool/string) from
80    /// ALL rows. Second pass coerces every value to the column's dominant type.
81    /// This ensures type consistency within each column.
82    pub async fn copy_from_csv(&self, collection: &str, csv_text: &str) -> NodeDbResult<u64> {
83        let mut lines_iter = csv_text.lines();
84        let header = lines_iter
85            .next()
86            .ok_or_else(|| NodeDbError::bad_request("CSV has no header"))?;
87        let columns: Vec<&str> = header.split(',').map(|s| s.trim()).collect();
88        let id_col = columns.iter().position(|c| *c == "id");
89
90        // Collect all data rows.
91        let rows: Vec<Vec<&str>> = lines_iter
92            .filter(|l| !l.trim().is_empty())
93            .map(|l| l.split(',').map(str::trim).collect())
94            .collect();
95
96        // Pass 1: infer column types. A column is "int" only if ALL values parse
97        // as int. "float" if all parse as float. "bool" if all are true/false.
98        // Otherwise "string".
99        let col_types: Vec<&str> = (0..columns.len())
100            .map(|ci| {
101                let all_int = rows
102                    .iter()
103                    .all(|r| r.get(ci).map(|v| v.parse::<i64>().is_ok()).unwrap_or(true));
104                if all_int {
105                    return "int";
106                }
107                let all_float = rows
108                    .iter()
109                    .all(|r| r.get(ci).map(|v| v.parse::<f64>().is_ok()).unwrap_or(true));
110                if all_float {
111                    return "float";
112                }
113                let all_bool = rows.iter().all(|r| {
114                    r.get(ci)
115                        .map(|v| *v == "true" || *v == "false")
116                        .unwrap_or(true)
117                });
118                if all_bool {
119                    return "bool";
120                }
121                "string"
122            })
123            .collect();
124
125        // Pass 2: build NDJSON with consistent types.
126        let mut ndjson = String::new();
127        for row in &rows {
128            let mut obj = serde_json::Map::new();
129            for (i, &col) in columns.iter().enumerate() {
130                if let Some(val) = row.get(i) {
131                    let json_val = match col_types[i] {
132                        "int" => val
133                            .parse::<i64>()
134                            .map(|n| serde_json::Value::Number(n.into()))
135                            .unwrap_or(serde_json::Value::Null),
136                        "float" => val
137                            .parse::<f64>()
138                            .ok()
139                            .and_then(serde_json::Number::from_f64)
140                            .map(serde_json::Value::Number)
141                            .unwrap_or(serde_json::Value::Null),
142                        "bool" => serde_json::Value::Bool(*val == "true"),
143                        _ => serde_json::Value::String((*val).to_owned()),
144                    };
145                    obj.insert(col.to_string(), json_val);
146                }
147            }
148            if id_col.is_none() {
149                obj.insert(
150                    "id".to_string(),
151                    serde_json::Value::String(nodedb_types::id_gen::uuid_v7()),
152                );
153            }
154            let row_json =
155                serde_json::to_string(&obj).map_err(|e| NodeDbError::serialization("json", e))?;
156            ndjson.push_str(&row_json);
157            ndjson.push('\n');
158        }
159        self.copy_from_ndjson(collection, &ndjson).await
160    }
161}
162
163/// Convert serde_json::Value to LoroValue for COPY FROM import.
164///
165/// Also used by the CRDT document-row ops, which receive their fields as a
166/// Control-Plane-built JSON object.
167pub(crate) fn json_to_loro(v: &serde_json::Value) -> loro::LoroValue {
168    match v {
169        serde_json::Value::Null => loro::LoroValue::Null,
170        serde_json::Value::Bool(b) => loro::LoroValue::Bool(*b),
171        serde_json::Value::Number(n) => {
172            if let Some(i) = n.as_i64() {
173                loro::LoroValue::I64(i)
174            } else if let Some(f) = n.as_f64() {
175                loro::LoroValue::Double(f)
176            } else {
177                loro::LoroValue::String(n.to_string().into())
178            }
179        }
180        serde_json::Value::String(s) => loro::LoroValue::String(s.clone().into()),
181        serde_json::Value::Array(arr) => {
182            loro::LoroValue::List(arr.iter().map(json_to_loro).collect::<Vec<_>>().into())
183        }
184        serde_json::Value::Object(map) => loro::LoroValue::Map(
185            map.iter()
186                .map(|(k, v)| (k.to_string(), json_to_loro(v)))
187                .collect::<std::collections::HashMap<_, _>>()
188                .into(),
189        ),
190    }
191}
192
193#[cfg(test)]
194mod tests {
195    use super::*;
196
197    #[test]
198    fn json_to_loro_roundtrip() {
199        let json = serde_json::json!({
200            "name": "Alice",
201            "age": 30,
202            "active": true,
203            "score": 99.5,
204            "tags": ["a", "b"]
205        });
206        let loro = json_to_loro(&json);
207        assert!(matches!(loro, loro::LoroValue::Map(_)));
208    }
209}
nodedb_lite/nodedb/collection/import.rs

nodedb_lite/nodedb/collection/
import.rs
