Fix review issues: explain plain SQL, strict node_type validation, config-embedded node checks

- explain.rs: Auto-wrap plain SQL input in df.explain() to prevent
  SELECT SELECT ... errors; detect DSL operators precisely to avoid
  misclassifying SQL with bitwise | or &
- types.rs: Add Durofut::ensure_strict() to reject unknown node_types
  at entrypoints; add validate_recursive() for full graph validation
  including config-embedded nodes (condition_node, extra_nodes)
- dsl.rs: df.start() now uses strict parsing + recursive validation;
  insert_nodes errors on invalid condition_node or extra_nodes
- lib.rs: SQL df.ensure_durofut() now rejects unknown node_type values
- Added E2E tests 27 (explain plain SQL) and 28 (invalid node_type)

Author: pinodeca

src/dsl.rs

@@ -477,7 +477,15 @@ pub fn signal(instance_id: &str, signal_name: &str, signal_data: default!(&str,
 /// Variables from df.vars are captured and passed to the orchestration.
 #[pg_extern(schema = "df")]
 pub fn start(fut: &str, label: default!(Option<&str>, "NULL")) -> String {
-    let durofut = Durofut::ensure(fut);
+    let durofut = match Durofut::ensure_strict(fut) {
+        Ok(d) => d,
+        Err(e) => pgrx::error!("Invalid durable function: {}", e),
+    };
+
+    // Validate the entire graph recursively before inserting
+    if let Err(e) = durofut.validate_recursive() {
+        pgrx::error!("Invalid durable function graph: {}", e);
+    }
     let instance_id = short_id();
 
     let label_sql = label
@@ -504,24 +512,40 @@ pub fn start(fut: &str, label: default!(Option<&str>, "NULL")) -> String {
                 // For IF/LOOP nodes: replace condition_node Durofut with ID
                 if node.node_type == "IF" || node.node_type == "LOOP" {
                     if let Some(cond_json) = config.get("condition_node") {
-                        if let Ok(cond_node) = serde_json::from_value::<Durofut>(cond_json.clone())
-                        {
-                            let cond_id = insert_nodes(&cond_node, instance_id);
-                            config["condition_node"] = serde_json::json!(cond_id);
+                        match serde_json::from_value::<Durofut>(cond_json.clone()) {
+                            Ok(cond_node) => {
+                                let cond_id = insert_nodes(&cond_node, instance_id);
+                                config["condition_node"] = serde_json::json!(cond_id);
+                            }
+                            Err(_) => {
+                                if cond_json.is_string() {
+                                    pgrx::error!(
+                                        "condition_node in {} must be a Durofut object, not a string ID",
+                                        node.node_type
+                                    );
+                                }
+                                // For other types, this will be caught by validate_recursive
+                            }
                         }
                     }
                 }
                 // For JOIN3 nodes: replace extra_nodes Durofuts with IDs
                 if node.node_type == "JOIN" {
                     if let Some(extras) = config.get("extra_nodes").and_then(|e| e.as_array()) {
-                        let extra_ids: Vec<String> = extras
-                            .iter()
-                            .filter_map(|extra_json| {
-                                serde_json::from_value::<Durofut>(extra_json.clone())
-                                    .ok()
-                                    .map(|n| insert_nodes(&n, instance_id))
-                            })
-                            .collect();
+                        let mut extra_ids: Vec<String> = Vec::new();
+                        for (i, extra_json) in extras.iter().enumerate() {
+                            match serde_json::from_value::<Durofut>(extra_json.clone()) {
+                                Ok(n) => {
+                                    extra_ids.push(insert_nodes(&n, instance_id));
+                                }
+                                Err(_) => {
+                                    pgrx::error!(
+                                        "extra_nodes[{}] in JOIN must be a Durofut object",
+                                        i
+                                    );
+                                }
+                            }
+                        }
                         if !extra_ids.is_empty() {
                             config["extra_nodes"] = serde_json::json!(extra_ids);
                         }

src/explain.rs

@@ -155,6 +155,45 @@ fn get_duroxide_instance_info(instance_id: &str) -> (String, Option<String>) {
     })
 }
 
+/// Check if input looks like a plain SQL statement (as opposed to a DSL expression).
+/// Returns true for inputs starting with SQL keywords like SELECT, INSERT, etc.
+fn looks_like_plain_sql(input: &str) -> bool {
+    let upper = input.trim_start().to_uppercase();
+    let sql_keywords = [
+        "SELECT ",
+        "INSERT ",
+        "UPDATE ",
+        "DELETE ",
+        "WITH ",
+        "CREATE ",
+        "ALTER ",
+        "DROP ",
+        "CALL ",
+        "DO ",
+        "TRUNCATE ",
+        "GRANT ",
+        "REVOKE ",
+        "EXPLAIN ",
+        "SELECT\n",
+        "INSERT\n",
+        "UPDATE\n",
+        "DELETE\n",
+        "WITH\n",
+    ];
+    sql_keywords.iter().any(|kw| upper.starts_with(kw))
+}
+
+/// Check if input looks like a DSL expression containing pg_durable operators or functions.
+/// DSL expressions contain operators like ~>, |=>, ?>, !>, @> or df.* function calls.
+fn looks_like_dsl_expression(input: &str) -> bool {
+    input.contains("~>")
+        || input.contains("|=>")
+        || input.contains("?>")
+        || input.contains("!>")
+        || input.contains("@>")
+        || input.contains("df.")
+}
+
 /// Explain a DSL expression without executing it
 fn explain_expression(expr: &str) -> String {
     use crate::types::Durofut;
@@ -168,7 +207,21 @@ fn explain_expression(expr: &str) -> String {
         return build_tree_visualization(&root_id, &nodes, false);
     }
 
-    // Not JSON - execute the expression to build the graph
+    // If it looks like plain SQL (not a DSL expression), wrap it directly as a SQL node
+    // This prevents the "SELECT SELECT ..." problem when users pass plain SQL to explain
+    if looks_like_plain_sql(expr) && !looks_like_dsl_expression(expr) {
+        let sql_node = Durofut {
+            node_type: "SQL".to_string(),
+            query: Some(expr.to_string()),
+            ..Default::default()
+        };
+        let mut nodes = HashMap::new();
+        let mut id_counter = 0;
+        let root_id = collect_nodes(&sql_node, &mut nodes, &mut id_counter);
+        return build_tree_visualization(&root_id, &nodes, false);
+    }
+
+    // DSL expression - evaluate it to build the graph
     let durofut_json: Result<Option<String>, _> = Spi::get_one(&format!("SELECT {expr}"));
 
     let root_json = match durofut_json {

src/lib.rs

@@ -174,14 +174,28 @@ END;
 $$ LANGUAGE plpgsql IMMUTABLE;
 
 -- Helper to ensure a value is a durofut (returns JSON string)
+-- Rejects JSON with unknown node_type values
 CREATE OR REPLACE FUNCTION df.ensure_durofut(val text) RETURNS text AS $$
+DECLARE
+    node_type_val text;
 BEGIN
     -- Try to parse as JSON to check if it's already a durofut
     BEGIN
-        IF (val::jsonb)->>'node_type' IS NOT NULL THEN
+        node_type_val := (val::jsonb)->>'node_type';
+        IF node_type_val IS NOT NULL THEN
+            -- Has a node_type - validate it
+            IF node_type_val NOT IN ('SQL', 'THEN', 'IF', 'JOIN', 'LOOP', 'BREAK', 'RACE', 'SLEEP', 'WAIT_SCHEDULE', 'HTTP', 'SIGNAL') THEN
+                RAISE EXCEPTION 'Unknown node_type ''%''. Valid types: SQL, THEN, IF, JOIN, LOOP, BREAK, RACE, SLEEP, WAIT_SCHEDULE, HTTP, SIGNAL', node_type_val;
+            END IF;
             RETURN val;
         END IF;
-    EXCEPTION WHEN OTHERS THEN
+    EXCEPTION WHEN invalid_text_representation THEN
+        -- Not valid JSON, treat as SQL
+        NULL;
+    WHEN raise_exception THEN
+        -- Re-raise our validation error
+        RAISE;
+    WHEN OTHERS THEN
         -- Not valid JSON, treat as SQL
         NULL;
     END;

src/types.rs

@@ -345,4 +345,88 @@ impl Durofut {
             },
         }
     }
+
+    /// Strict version of ensure - rejects JSON with unknown node_type instead of wrapping as SQL.
+    /// Used by df.start() and other entrypoints where invalid node types should be caught early.
+    pub fn ensure_strict(s: &str) -> Result<Self, String> {
+        match serde_json::from_str::<Durofut>(s) {
+            Ok(d) => {
+                if VALID_NODE_TYPES.contains(&d.node_type.as_str()) {
+                    Ok(d)
+                } else {
+                    Err(format!(
+                        "Unknown node_type '{}'. Valid types: {}",
+                        d.node_type,
+                        VALID_NODE_TYPES.join(", ")
+                    ))
+                }
+            }
+            Err(_) => {
+                // Not valid Durofut JSON - try to parse as generic JSON to check for node_type
+                if let Ok(val) = serde_json::from_str::<serde_json::Value>(s) {
+                    if let Some(nt) = val.get("node_type").and_then(|v| v.as_str()) {
+                        return Err(format!(
+                            "Unknown node_type '{}'. Valid types: {}",
+                            nt,
+                            VALID_NODE_TYPES.join(", ")
+                        ));
+                    }
+                }
+                // Not JSON at all or no node_type field - treat as SQL
+                Ok(Durofut {
+                    node_type: "SQL".to_string(),
+                    query: Some(s.to_string()),
+                    ..Default::default()
+                })
+            }
+        }
+    }
+
+    /// Validate a Durofut node and all its children have valid node_types.
+    /// Used during insertion in df.start() to catch invalid nested nodes.
+    pub fn validate_recursive(&self) -> Result<(), String> {
+        if !VALID_NODE_TYPES.contains(&self.node_type.as_str()) {
+            return Err(format!(
+                "Unknown node_type '{}'. Valid types: {}",
+                self.node_type,
+                VALID_NODE_TYPES.join(", ")
+            ));
+        }
+        if let Some(ref left) = self.left_node {
+            left.validate_recursive()?;
+        }
+        if let Some(ref right) = self.right_node {
+            right.validate_recursive()?;
+        }
+        // Validate config-embedded nodes (condition_node, extra_nodes)
+        if let Some(ref query_str) = self.query {
+            if let Ok(config) = serde_json::from_str::<serde_json::Value>(query_str) {
+                if let Some(cond) = config.get("condition_node") {
+                    if let Ok(cond_node) = serde_json::from_value::<Durofut>(cond.clone()) {
+                        cond_node.validate_recursive()?;
+                    } else if cond.is_string()
+                        && (self.node_type == "IF" || self.node_type == "LOOP")
+                    {
+                        return Err(format!(
+                            "condition_node in {} must be a Durofut object, not a string ID",
+                            self.node_type
+                        ));
+                    }
+                }
+                if let Some(extras) = config.get("extra_nodes").and_then(|e| e.as_array()) {
+                    for (i, extra) in extras.iter().enumerate() {
+                        if let Ok(extra_node) = serde_json::from_value::<Durofut>(extra.clone()) {
+                            extra_node.validate_recursive()?;
+                        } else {
+                            return Err(format!(
+                                "extra_nodes[{}] in {} must be a Durofut object",
+                                i, self.node_type
+                            ));
+                        }
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
 }

tests/e2e/sql/27_explain_plain_sql.sql

@@ -0,0 +1,21 @@
+-- Test: Explain on plain SQL input (auto-wrap)
+-- Expected: df.explain('SELECT 1') produces a SQL node visualization
+
+DO $body$
+DECLARE
+    explain_output TEXT;
+BEGIN
+    SELECT df.explain('SELECT 1') INTO explain_output;
+
+    IF explain_output IS NULL OR explain_output = '' THEN
+        RAISE EXCEPTION 'TEST FAILED: explain returned empty output';
+    END IF;
+
+    IF explain_output NOT LIKE '%SQL:%' OR explain_output NOT LIKE '%SELECT 1%' THEN
+        RAISE EXCEPTION 'TEST FAILED: explain should show SQL: SELECT 1, got: %', explain_output;
+    END IF;
+
+    RAISE NOTICE 'TEST PASSED: explain plain SQL';
+END $body$;
+
+SELECT 'TEST PASSED' AS result;

tests/e2e/sql/28_invalid_node_type.sql

@@ -0,0 +1,25 @@
+-- Test: Reject invalid Durofut JSON node_type
+-- Expected: df.start('{"node_type":"NOT_A_NODE"}') raises an error
+
+DO $body$
+BEGIN
+    BEGIN
+        PERFORM df.start('{"node_type":"NOT_A_NODE"}');
+        RAISE EXCEPTION 'TEST FAILED: df.start should have rejected invalid node_type';
+    EXCEPTION WHEN OTHERS THEN
+        -- ok
+        RAISE NOTICE 'Caught expected error: %', SQLERRM;
+    END;
+
+    BEGIN
+        PERFORM df.explain('{"node_type":"NOT_A_NODE"}');
+        -- explain returns a string; it should contain our error text, not crash.
+        -- If it returned empty, that would be suspicious.
+    EXCEPTION WHEN OTHERS THEN
+        RAISE EXCEPTION 'TEST FAILED: df.explain should not raise on invalid node_type';
+    END;
+
+    RAISE NOTICE 'TEST PASSED: invalid node_type handling';
+END $body$;
+
+SELECT 'TEST PASSED' AS result;