1use crate::expr::{evaluate, ExprFailure, ExprFailureCode};
18use crate::plan::RelationKind;
19use crate::value::Value;
20use serde_json::{json, Value as J};
21
22pub type Expr = J;
24pub type R = Result<Value, ExprFailure>;
26
27pub fn r#ref(path: &[&str], scope: &[(String, Value)]) -> R {
32 evaluate(&json!({ "ref": path }), scope)
33}
34pub fn ref_opt(path: &[&str], scope: &[(String, Value)]) -> R {
36 evaluate(&json!({ "refOpt": path }), scope)
37}
38pub fn int_lit(literal: &str, scope: &[(String, Value)]) -> R {
40 evaluate(&json!({ "int": literal }), scope)
41}
42pub fn float_lit(n: f64, scope: &[(String, Value)]) -> R {
44 evaluate(&json!({ "float": n }), scope)
45}
46pub fn obj(fields: &J, scope: &[(String, Value)]) -> R {
48 evaluate(&json!({ "obj": fields }), scope)
49}
50pub fn arr(elems: &[J], scope: &[(String, Value)]) -> R {
52 evaluate(&json!({ "arr": elems }), scope)
53}
54
55pub fn add(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
57 evaluate(&json!({ "add": [a, b] }), scope)
58}
59pub fn sub(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
61 evaluate(&json!({ "sub": [a, b] }), scope)
62}
63pub fn mul(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
65 evaluate(&json!({ "mul": [a, b] }), scope)
66}
67pub fn neg(a: &Expr, scope: &[(String, Value)]) -> R {
69 evaluate(&json!({ "neg": [a] }), scope)
70}
71pub fn div(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
73 evaluate(&json!({ "div": [a, b] }), scope)
74}
75pub fn r#mod(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
77 evaluate(&json!({ "mod": [a, b] }), scope)
78}
79
80pub fn concat(parts: &[J], scope: &[(String, Value)]) -> R {
82 evaluate(&json!({ "concat": parts }), scope)
83}
84
85pub fn eq(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
87 evaluate(&json!({ "eq": [a, b] }), scope)
88}
89pub fn ne(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
91 evaluate(&json!({ "ne": [a, b] }), scope)
92}
93pub fn lt(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
95 evaluate(&json!({ "lt": [a, b] }), scope)
96}
97pub fn le(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
99 evaluate(&json!({ "le": [a, b] }), scope)
100}
101pub fn gt(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
103 evaluate(&json!({ "gt": [a, b] }), scope)
104}
105pub fn ge(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
107 evaluate(&json!({ "ge": [a, b] }), scope)
108}
109
110pub fn and(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
112 evaluate(&json!({ "and": [a, b] }), scope)
113}
114pub fn or(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
116 evaluate(&json!({ "or": [a, b] }), scope)
117}
118pub fn not(a: &Expr, scope: &[(String, Value)]) -> R {
120 evaluate(&json!({ "not": [a] }), scope)
121}
122pub fn coalesce(a: &Expr, b: &Expr, scope: &[(String, Value)]) -> R {
124 evaluate(&json!({ "coalesce": [a, b] }), scope)
125}
126pub fn cond(c: &Expr, then: &Expr, els: &Expr, scope: &[(String, Value)]) -> R {
128 evaluate(&json!({ "cond": [c, then, els] }), scope)
129}
130pub fn len(a: &Expr, scope: &[(String, Value)]) -> R {
132 evaluate(&json!({ "len": [a] }), scope)
133}
134
135pub fn eval_expr(node: &Expr, scope: &[(String, Value)]) -> R {
137 evaluate(node, scope)
138}
139
140fn expr_fail(code: ExprFailureCode, message: String) -> ExprFailure {
152 ExprFailure { code, message }
153}
154
155pub fn ref_native(path: &[&str], scope: &[(String, Value)]) -> R {
158 ref_walk(path, scope, false)
159}
160
161pub fn ref_opt_native(path: &[&str], scope: &[(String, Value)]) -> R {
163 ref_walk(path, scope, true)
164}
165
166fn ref_walk(path: &[&str], scope: &[(String, Value)], optional: bool) -> R {
169 let op = if optional { "refOpt" } else { "ref" };
170 if path.is_empty() {
171 return Err(expr_fail(
172 ExprFailureCode::InvalidNode,
173 format!("{op} expects a non-empty string path"),
174 ));
175 }
176 let head = path[0];
177 let mut cur: Value = match scope.iter().find(|(k, _)| k == head) {
178 Some((_, v)) => v.clone(),
179 None => {
180 return Err(expr_fail(
181 ExprFailureCode::UnknownBinding,
182 format!("unknown binding: {head}"),
183 ))
184 }
185 };
186 for &seg in &path[1..] {
187 match cur {
188 Value::Null => {
189 if optional {
190 return Ok(Value::Null);
191 }
192 return Err(expr_fail(
193 ExprFailureCode::NullRef,
194 format!("null intermediate at .{seg} (use ?.)"),
195 ));
196 }
197 Value::Obj(ref pairs) => match pairs.iter().find(|(k, _)| k == seg) {
198 Some((_, v)) => cur = v.clone(),
199 None => {
200 return Err(expr_fail(
201 ExprFailureCode::MissingProp,
202 format!("missing property .{seg}"),
203 ))
204 }
205 },
206 ref other => {
207 return Err(expr_fail(
208 ExprFailureCode::TypeMismatch,
209 format!("cannot access .{seg} on {}", other.type_name()),
210 ))
211 }
212 }
213 }
214 Ok(cur)
215}
216
217pub fn concat_native(parts: &[Value]) -> R {
222 if parts.len() < 2 {
223 return Err(expr_fail(
224 ExprFailureCode::InvalidNode,
225 format!("concat expects >= 2 args, got {}", parts.len()),
226 ));
227 }
228 let mut s = String::new();
229 for p in parts {
230 match p {
231 Value::Str(part) => s.push_str(part),
232 other => {
233 return Err(expr_fail(
234 ExprFailureCode::TypeMismatch,
235 format!(
236 "concat: strings only (got {}; no implicit toString)",
237 other.type_name()
238 ),
239 ))
240 }
241 }
242 }
243 Ok(Value::Str(s))
244}
245
246pub fn map_with_concurrency<T, U, F>(items: &[T], concurrency: i64, worker: F) -> Vec<U>
256where
257 T: Sync,
258 U: Send,
259 F: Fn(&T, usize) -> U + Sync,
260{
261 let n = items.len();
262 let limit = concurrency.max(1) as usize;
263 if limit <= 1 || n <= 1 {
264 return items
265 .iter()
266 .enumerate()
267 .map(|(i, it)| worker(it, i))
268 .collect();
269 }
270 let mut slots: Vec<Option<U>> = (0..n).map(|_| None).collect();
272 let cursor = std::sync::atomic::AtomicUsize::new(0);
273 let workers = limit.min(n);
274 let worker_ref = &worker;
275 let items_ref = items;
276 {
277 struct SlotsPtr<U>(*mut Option<U>);
279 unsafe impl<U: Send> Sync for SlotsPtr<U> {}
280 let base = SlotsPtr(slots.as_mut_ptr());
281 std::thread::scope(|s| {
282 for _ in 0..workers {
283 let cursor = &cursor;
284 let base = &base;
285 s.spawn(move || loop {
286 let i = cursor.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
287 if i >= n {
288 break;
289 }
290 let out = worker_ref(&items_ref[i], i);
291 unsafe {
294 *base.0.add(i) = Some(out);
295 }
296 });
297 }
298 });
299 }
300 slots
301 .into_iter()
302 .map(|o| o.expect("all slots filled"))
303 .collect()
304}
305
306pub fn unproduced_value(kind: Option<RelationKind>) -> Value {
309 match kind {
310 Some(RelationKind::Connection) => skip_connection(),
311 _ => Value::Null,
312 }
313}
314pub fn skip_single() -> Value {
316 Value::Null
317}
318pub fn skip_connection() -> Value {
320 Value::Obj(vec![
321 ("items".to_string(), Value::Arr(vec![])),
322 ("cursor".to_string(), Value::Null),
323 ])
324}
325
326pub fn hydrate_into(
333 over: &[Value],
334 key: &str,
335 kept_idx: &[usize],
336 values: &[Value],
337) -> Result<Vec<Value>, String> {
338 let mut augmented: Vec<Value> = Vec::with_capacity(over.len());
339 let mut k = 0usize;
340 for (i, el) in over.iter().enumerate() {
341 if k < kept_idx.len() && kept_idx[k] == i {
342 match el {
343 Value::Obj(pairs) => {
344 let mut next: Vec<(String, Value)> = pairs.clone();
345 if let Some(slot) = next.iter_mut().find(|(kk, _)| kk == key) {
347 slot.1 = values[k].clone();
348 } else {
349 next.push((key.to_string(), values[k].clone()));
350 }
351 augmented.push(Value::Obj(next));
352 }
353 _ => {
354 return Err(format!(
355 "hydrate_into: element {i} is not an object (into requires object elements)"
356 ))
357 }
358 }
359 k += 1;
360 } else {
361 augmented.push(el.clone());
362 }
363 }
364 Ok(augmented)
365}