use std::path::PathBuf;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use crate::pipeline::nodes::shared::util::{metadata_scope, resolve_path};
use crate::pipeline::model::NodeCapability;
use crate::pipeline::model::{DslFlag, DslFlagKind, LayoutItem, NodeFieldDef, NodeFieldType};
use crate::pipeline::{
NodeDefinition, PipelineError,
nodes::{NodeExecutionInput, NodeExecutionOutput, NodeHandler},
};
use crate::platform::sekejap::{
self, StructuredInsertEdge, StructuredInsertRecord, StructuredWriteMode,
};
pub const NODE_KIND: &str = "n.sekejap.insert";
pub const INPUT_PIN_IN: &str = "in";
pub const OUTPUT_PIN_OUT: &str = "out";
pub fn definition() -> NodeDefinition {
NodeDefinition {
kind: NODE_KIND.to_string(),
capabilities: vec![NodeCapability::Database],
title: "Sekejap Insert".to_string(),
description: "Bulk-insert records (and optional graph edges) into a Sekejap collection, as one commit. Reads an array at \
`--records-path` (default `records`), each `{ key, fields: { … } }`, and writes them typed against the collection's \
declared columns into `--target` — the collection must already exist (`CREATE TABLE`), a `VECTOR(n)` field must have \
exactly n numbers, and an edge's two endpoints must exist, in this batch or before it. An edge whose `type` is an \
edge table's label is written into that table, its `fields` as the table's columns; the target itself must be a \
table of rows. Answers \
`{ inserted_records, inserted_edges, … }` — the payload is replaced. For one row from a form use `sekejap.query … INSERT`; \
this node is for imports and seeds, up to `--max-records` (default 1000) per run."
.to_string(),
input_schema: json!({
"type": "object",
"description": "Payload containing records and optional native edges. Each record must include a key and fields object.",
"properties": {
"records": {
"type": "array",
"items": {
"type": "object",
"required": ["key", "fields"],
"properties": {
"key": { "type": ["string", "number"] },
"fields": { "type": "object" }
}
}
},
"edges": {
"type": "array",
"items": {
"type": "object",
"required": ["from", "type", "to"],
"properties": {
"from": { "type": "object" },
"type": { "type": "string" },
"to": { "type": "object" },
"fields": { "type": "object" },
"strength": { "type": "number" }
}
}
}
}
}),
output_schema: json!({
"type": "object",
"properties": {
"inserted_records": { "type": "integer" },
"inserted_edges": { "type": "integer" },
"optimized_fields": { "type": "array", "items": { "type": "string" } },
"field_dimensions": { "type": "object" },
"duration_ms": { "type": "integer" }
}
}),
input_pins: vec![INPUT_PIN_IN.to_string()],
output_pins: vec![OUTPUT_PIN_OUT.to_string()],
script_available: false,
script_bridge: None,
config_schema: Default::default(),
dsl_flags: vec![
DslFlag { flag: "--target".to_string(), config_key: "target".to_string(), description: "Sekejap collection/type for records.".to_string(), kind: DslFlagKind::Scalar, required: true },
DslFlag { flag: "--records-path".to_string(), config_key: "records_path".to_string(), description: "Dot path to the input records array. Default: records.".to_string(), kind: DslFlagKind::Scalar, required: false },
DslFlag { flag: "--edges-path".to_string(), config_key: "edges_path".to_string(), description: "Dot path to the optional native edges array. Default: edges.".to_string(), kind: DslFlagKind::Scalar, required: false },
DslFlag { flag: "--key-path".to_string(), config_key: "key_path".to_string(), description: "Dot path inside each record for the record key. Default: key.".to_string(), kind: DslFlagKind::Scalar, required: false },
DslFlag { flag: "--max-records".to_string(), config_key: "max_records".to_string(), description: "Maximum records accepted in one execution. Default: 1000.".to_string(), kind: DslFlagKind::Scalar, required: false },
DslFlag { flag: "--max-edges".to_string(), config_key: "max_edges".to_string(), description: "Maximum edges accepted in one execution. Default: 1000.".to_string(), kind: DslFlagKind::Scalar, required: false },
],
fields: vec![
NodeFieldDef { name: "target".to_string(), label: "Target".to_string(), field_type: NodeFieldType::Text, help: Some("Sekejap collection/type for inserted records.".to_string()), ..Default::default() },
NodeFieldDef { name: "records_path".to_string(), label: "Records Path".to_string(), field_type: NodeFieldType::Text, default_value: Some(json!("records")), help: Some("Dot path to records in the input payload.".to_string()), ..Default::default() },
NodeFieldDef { name: "edges_path".to_string(), label: "Edges Path".to_string(), field_type: NodeFieldType::Text, default_value: Some(json!("edges")), help: Some("Dot path to optional native edges in the input payload.".to_string()), ..Default::default() },
NodeFieldDef { name: "key_path".to_string(), label: "Key Path".to_string(), field_type: NodeFieldType::Text, default_value: Some(json!("key")), help: Some("Dot path inside each record for the record key.".to_string()), ..Default::default() },
NodeFieldDef { name: "max_records".to_string(), label: "Max Records".to_string(), field_type: NodeFieldType::Number, default_value: Some(json!(1000)), help: Some("Maximum records accepted in one execution.".to_string()), ..Default::default() },
NodeFieldDef { name: "max_edges".to_string(), label: "Max Edges".to_string(), field_type: NodeFieldType::Number, default_value: Some(json!(1000)), help: Some("Maximum edges accepted in one execution.".to_string()), ..Default::default() },
],
layout: vec![
LayoutItem::Field("target".to_string()),
LayoutItem::Row { row: vec![LayoutItem::Field("records_path".to_string()), LayoutItem::Field("edges_path".to_string())] },
LayoutItem::Row { row: vec![LayoutItem::Field("key_path".to_string()), LayoutItem::Field("max_records".to_string()), LayoutItem::Field("max_edges".to_string())] },
],
ai_tool: crate::pipeline::model::NodeAiToolDefinition {
registered: true,
tool_name: "sekejap_insert".to_string(),
tool_description: "Bulk insert records and native Sekejap edges without generating SQL strings. Args: target, records_path, edges_path, key_path, max_records, max_edges. Vector fields are optimized automatically.".to_string(),
tool_input_schema: json!({
"type": "object",
"properties": {
"target": { "type": "string" },
"records_path": { "type": "string" },
"edges_path": { "type": "string" },
"key_path": { "type": "string" },
"max_records": { "type": "integer" },
"max_edges": { "type": "integer" }
},
"required": ["target"]
}),
},
examples: vec![
crate::pipeline::model::NodeExample::dsl("Seed from a prepared array", r#"sekejap.insert --target products --records-path items"#)
.input(serde_json::json!({ "items": [{ "key": "sku-1", "fields": { "name": "Mug", "price": 12 } }] }))
.output(serde_json::json!({ "inserted_records": 1, "inserted_edges": 0 })),
],
..Default::default()
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Config {
pub target: String,
#[serde(default = "default_records_path")]
pub records_path: String,
#[serde(default = "default_edges_path")]
pub edges_path: String,
#[serde(default = "default_key_path")]
pub key_path: String,
#[serde(default = "default_max_records")]
pub max_records: usize,
#[serde(default = "default_max_edges")]
pub max_edges: usize,
}
fn default_records_path() -> String {
"records".to_string()
}
fn default_edges_path() -> String {
"edges".to_string()
}
fn default_key_path() -> String {
"key".to_string()
}
fn default_max_records() -> usize {
1000
}
fn default_max_edges() -> usize {
1000
}
pub struct Node {
config: Config,
data_root: PathBuf,
}
impl Node {
pub fn new(config: Config, data_root: PathBuf) -> Result<Self, PipelineError> {
if config.target.trim().is_empty() {
return Err(PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_CONFIG",
"target must not be empty",
));
}
if config.records_path.trim().is_empty() {
return Err(PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_CONFIG",
"records_path must not be empty",
));
}
if config.edges_path.trim().is_empty() {
return Err(PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_CONFIG",
"edges_path must not be empty",
));
}
if config.key_path.trim().is_empty() {
return Err(PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_CONFIG",
"key_path must not be empty",
));
}
if config.max_records == 0 {
return Err(PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_CONFIG",
"max_records must be greater than zero",
));
}
if config.max_edges == 0 {
return Err(PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_CONFIG",
"max_edges must be greater than zero",
));
}
Ok(Self { config, data_root })
}
}
#[async_trait]
impl NodeHandler for Node {
fn kind(&self) -> &'static str {
NODE_KIND
}
fn input_pins(&self) -> &'static [&'static str] {
&[INPUT_PIN_IN]
}
fn output_pins(&self) -> &'static [&'static str] {
&[OUTPUT_PIN_OUT]
}
async fn execute_async(
&self,
input: NodeExecutionInput,
) -> Result<NodeExecutionOutput, PipelineError> {
let (owner, project, _pipeline, _request_id) = metadata_scope(&input.metadata)?;
let records_array =
optional_array_at_path(&input.payload, self.config.records_path.trim())?;
let edges_array = optional_array_at_path(&input.payload, self.config.edges_path.trim())?;
if records_array.len() > self.config.max_records {
return Err(PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_LIMIT",
format!(
"batch has {} records, max_records is {}",
records_array.len(),
self.config.max_records
),
));
}
if edges_array.len() > self.config.max_edges {
return Err(PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_LIMIT",
format!(
"batch has {} edges, max_edges is {}",
edges_array.len(),
self.config.max_edges
),
));
}
if records_array.is_empty() && edges_array.is_empty() {
return Err(PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_INPUT",
"insert payload must include at least one record or edge",
));
}
let mut records = Vec::with_capacity(records_array.len());
for (index, record) in records_array.iter().enumerate() {
let key =
scalar_key(resolve_path(record, self.config.key_path.trim())).ok_or_else(|| {
PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_INPUT",
format!(
"record {index} missing scalar key at '{}'",
self.config.key_path
),
)
})?;
let fields = record
.get("fields")
.and_then(Value::as_object)
.cloned()
.ok_or_else(|| {
PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_INPUT",
format!("record {index} must include a fields object"),
)
})?;
records.push(StructuredInsertRecord { key, fields });
}
let mut edges = Vec::with_capacity(edges_array.len());
for (index, edge) in edges_array.iter().enumerate() {
edges.push(parse_edge(index, edge)?);
}
let data_root = self.data_root.clone();
let owner = owner.to_string();
let project = project.to_string();
let target = self.config.target.clone();
let record_count = records.len();
let edge_count = edges.len();
let result = tokio::task::spawn_blocking(move || {
sekejap::bulk_insert(
&data_root,
&owner,
&project,
&target,
records,
edges,
StructuredWriteMode::Insert,
)
})
.await
.map_err(|err| {
PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_JOIN",
format!("sekejap insert task failed: {err}"),
)
})?
.map_err(|err| PipelineError::new("FW_NODE_SEKEJAP_INSERT", err.to_string()))?;
Ok(NodeExecutionOutput {
output_pins: vec![OUTPUT_PIN_OUT.to_string()],
payload: json!({
"inserted_records": record_count,
"inserted_edges": edge_count,
"affected_rows": result.affected_rows,
"optimized_fields": result.optimized_fields,
"field_dimensions": result.field_dimensions,
"duration_ms": result.duration_ms,
}),
trace: vec![
format!("node_kind={NODE_KIND}"),
format!("inserted_records={record_count}"),
format!("inserted_edges={edge_count}"),
format!("optimized_fields={}", result.optimized_fields.len()),
],
})
}
}
fn optional_array_at_path(payload: &Value, path: &str) -> Result<Vec<Value>, PipelineError> {
let Some(value) = resolve_path(payload, path) else {
return Ok(Vec::new());
};
value.as_array().cloned().ok_or_else(|| {
PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_INPUT",
format!("{path} must resolve to an array"),
)
})
}
fn parse_edge(index: usize, edge: &Value) -> Result<StructuredInsertEdge, PipelineError> {
let edge_type = edge
.get("type")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| {
PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_INPUT",
format!("edge {index} missing non-empty type"),
)
})?
.to_string();
let from = edge_endpoint(index, "from", edge.get("from"))?;
let to = edge_endpoint(index, "to", edge.get("to"))?;
let fields = edge
.get("fields")
.and_then(Value::as_object)
.cloned()
.unwrap_or_default();
let strength = edge.get("strength").and_then(Value::as_f64).unwrap_or(1.0) as f32;
Ok(StructuredInsertEdge {
from_target: from.0,
from_key: from.1,
edge_type,
to_target: to.0,
to_key: to.1,
fields,
strength,
})
}
fn edge_endpoint(
index: usize,
side: &str,
value: Option<&Value>,
) -> Result<(String, String), PipelineError> {
let value = value.and_then(Value::as_object).ok_or_else(|| {
PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_INPUT",
format!("edge {index} {side} must be an object"),
)
})?;
let target = value
.get("target")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| {
PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_INPUT",
format!("edge {index} {side}.target must be a non-empty string"),
)
})?
.to_string();
let key = scalar_key(value.get("key")).ok_or_else(|| {
PipelineError::new(
"FW_NODE_SEKEJAP_INSERT_INPUT",
format!("edge {index} {side}.key must be a string or number"),
)
})?;
Ok((target, key))
}
fn scalar_key(value: Option<&Value>) -> Option<String> {
match value? {
Value::String(s) => Some(s.clone()),
Value::Number(n) => Some(n.to_string()),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[tokio::test]
async fn writes_edges_whose_type_is_an_edge_table_into_that_table() {
let tmp = tempfile::tempdir().expect("tempdir");
for sql in [
"CREATE TABLE people (_key TEXT PRIMARY KEY, name TEXT)",
"CREATE TABLE knows (src TEXT REFERENCES people, dst TEXT REFERENCES people, since INT, PRIMARY KEY (src, dst))",
"CREATE PROPERTY GRAPH social VERTEX TABLES (people) EDGE TABLES (knows SOURCE KEY (src) REFERENCES people (_key) DESTINATION KEY (dst) REFERENCES people (_key))",
] {
crate::platform::sekejap::execute_sql(tmp.path(), "alice", "demo", sql, &[], 10, false)
.expect(sql);
}
let node = Node::new(
Config {
target: "people".to_string(),
records_path: "records".to_string(),
edges_path: "edges".to_string(),
key_path: "key".to_string(),
max_records: 10,
max_edges: 10,
},
tmp.path().to_path_buf(),
)
.expect("node");
let output = node
.execute_async(NodeExecutionInput {
node_id: "insert".to_string(),
input_pin: "in".to_string(),
payload: json!({
"records": [
{ "key": "ann", "fields": { "name": "Ann" } },
{ "key": "bo", "fields": { "name": "Bo" } }
],
"edges": [
{
"from": { "target": "people", "key": "ann" },
"type": "knows",
"to": { "target": "people", "key": "bo" },
"fields": { "since": 2020 }
}
]
}),
metadata: json!({ "owner": "alice", "project": "demo", "pipeline": "t", "request_id": "r" }),
bus: None,
})
.await
.expect("execute");
assert_eq!(output.payload["inserted_records"], json!(2));
assert_eq!(output.payload["inserted_edges"], json!(1));
let walked = crate::platform::sekejap::execute_sql(
tmp.path(),
"alice",
"demo",
"SELECT name, since FROM GRAPH_TABLE (social MATCH (a WHERE a._key = 'ann')-[e:knows]->(b) RETURN b.name AS name, e.since AS since)",
&[],
10,
true,
)
.expect("walk");
assert_eq!(walked.rows, vec![vec![json!("Bo"), json!(2020)]]);
}
#[tokio::test]
async fn executes_records_and_native_edges_from_insert_contract() {
let tmp = tempfile::tempdir().expect("tempdir");
for sql in [
"CREATE TABLE documents (_key TEXT PRIMARY KEY, title TEXT)",
"CREATE TABLE authors (_key TEXT PRIMARY KEY, name TEXT)",
"INSERT INTO authors (_key, name) VALUES ('author:1', 'Ann')",
] {
crate::platform::sekejap::execute_sql(tmp.path(), "alice", "demo", sql, &[], 10, false)
.expect(sql);
}
let node = Node::new(
Config {
target: "documents".to_string(),
records_path: "records".to_string(),
edges_path: "edges".to_string(),
key_path: "key".to_string(),
max_records: 10,
max_edges: 10,
},
tmp.path().to_path_buf(),
)
.expect("node");
let output = node
.execute_async(NodeExecutionInput {
node_id: "insert".to_string(),
input_pin: "in".to_string(),
payload: json!({
"records": [
{
"key": "doc:1",
"fields": {
"title": "One"
}
}
],
"edges": [
{
"from": { "target": "authors", "key": "author:1" },
"type": "wrote",
"to": { "target": "documents", "key": "doc:1" },
"fields": { "year": 2026 },
"strength": 1.0
}
]
}),
metadata: json!({
"owner": "alice",
"project": "demo",
"pipeline": "insert-test",
"request_id": "req-1"
}),
bus: None,
})
.await
.expect("execute");
assert_eq!(output.payload["inserted_records"], json!(1));
assert_eq!(output.payload["inserted_edges"], json!(1));
assert_eq!(output.payload["affected_rows"], json!(2));
}
}