Add triggers property and validation to n8n schema

Co-authored-by: johndoe6345789 <224850594+johndoe6345789@users.noreply.github.com>

ROADMAP.md

@@ -158,6 +158,12 @@
 "type": "array",
 "items": { "$ref": "#/$defs/credentialBinding" },
 "default": []
+},
+"triggers": {
+"description": "Optional explicit trigger declarations for event-driven workflows.",
+"type": "array",
+"default": [],
+"items": { "$ref": "#/$defs/trigger" }
 }
 },
 "$defs": {
@@ -399,6 +405,25 @@
 "minimum": 0
 }
 }
+},
+"trigger": {
+"type": "object",
+"additionalProperties": false,
+"required": ["nodeId", "kind"],
+"properties": {
+"nodeId": { "type": "string", "minLength": 1 },
+"kind": {
+"type": "string",
+"enum": ["webhook", "schedule", "queue", "email", "poll", "manual", "other"]
+},
+"enabled": { "type": "boolean", "default": true },
+"meta": {
+"description": "Trigger-kind-specific metadata for routing/registration.",
+"type": "object",
+"additionalProperties": true,
+"default": {}
+}
+}
 }
 }
 }

backend/autometabuilder/workflow/n8n_schema.py

@@ -53,6 +53,29 @@ class N8NNode:
         return True
 
 
+class N8NTrigger:
+    """N8N workflow trigger specification."""
+    
+    VALID_KINDS = ["webhook", "schedule", "queue", "email", "poll", "manual", "other"]
+    
+    @staticmethod
+    def validate(value: Any) -> bool:
+        if not isinstance(value, dict):
+            return False
+        required = ["nodeId", "kind"]
+        if not all(key in value for key in required):
+            return False
+        if not isinstance(value["nodeId"], str) or not value["nodeId"]:
+            return False
+        if not isinstance(value["kind"], str) or value["kind"] not in N8NTrigger.VALID_KINDS:
+            return False
+        if "enabled" in value and not isinstance(value["enabled"], bool):
+            return False
+        if "meta" in value and not isinstance(value["meta"], dict):
+            return False
+        return True
+
+
 class N8NWorkflow:
     """N8N workflow specification."""
     
@@ -69,4 +92,12 @@ class N8NWorkflow:
             return False
         if not isinstance(value["connections"], dict):
             return False
-        return all(N8NNode.validate(node) for node in value["nodes"])
+        if not all(N8NNode.validate(node) for node in value["nodes"]):
+            return False
+        # Validate triggers array if present
+        if "triggers" in value:
+            if not isinstance(value["triggers"], list):
+                return False
+            if not all(N8NTrigger.validate(trigger) for trigger in value["triggers"]):
+                return False
+        return True

backend/tests/test_n8n_schema.py

@@ -1,7 +1,7 @@
 """Tests for n8n workflow schema validation."""
 import pytest
 
-from autometabuilder.workflow.n8n_schema import N8NNode, N8NPosition, N8NWorkflow
+from autometabuilder.workflow.n8n_schema import N8NNode, N8NPosition, N8NTrigger, N8NWorkflow
 
 
 def test_n8n_position_validation():
@@ -71,3 +71,123 @@ def test_n8n_workflow_validation():
     invalid_workflow = valid_workflow.copy()
     invalid_workflow["nodes"] = [{"id": "bad"}]
     assert not N8NWorkflow.validate(invalid_workflow)
+
+
+def test_n8n_trigger_validation():
+    """Test trigger validation."""
+    valid_trigger = {
+        "nodeId": "webhook-node-1",
+        "kind": "webhook",
+        "enabled": True,
+        "meta": {
+            "path": "/api/webhook",
+            "method": "POST"
+        }
+    }
+    assert N8NTrigger.validate(valid_trigger)
+    
+    # Minimal valid trigger
+    minimal_trigger = {
+        "nodeId": "schedule-node",
+        "kind": "schedule"
+    }
+    assert N8NTrigger.validate(minimal_trigger)
+    
+    # Test all valid kinds
+    for kind in ["webhook", "schedule", "queue", "email", "poll", "manual", "other"]:
+        trigger = {"nodeId": "node-1", "kind": kind}
+        assert N8NTrigger.validate(trigger)
+    
+    # Missing required fields
+    assert not N8NTrigger.validate({})
+    assert not N8NTrigger.validate({"nodeId": "node-1"})
+    assert not N8NTrigger.validate({"kind": "webhook"})
+    
+    # Invalid nodeId
+    invalid_trigger = valid_trigger.copy()
+    invalid_trigger["nodeId"] = ""
+    assert not N8NTrigger.validate(invalid_trigger)
+    
+    invalid_trigger = valid_trigger.copy()
+    invalid_trigger["nodeId"] = 123
+    assert not N8NTrigger.validate(invalid_trigger)
+    
+    # Invalid kind
+    invalid_trigger = valid_trigger.copy()
+    invalid_trigger["kind"] = "invalid_kind"
+    assert not N8NTrigger.validate(invalid_trigger)
+    
+    # Invalid enabled
+    invalid_trigger = valid_trigger.copy()
+    invalid_trigger["enabled"] = "true"
+    assert not N8NTrigger.validate(invalid_trigger)
+    
+    # Invalid meta
+    invalid_trigger = valid_trigger.copy()
+    invalid_trigger["meta"] = "not a dict"
+    assert not N8NTrigger.validate(invalid_trigger)
+
+
+def test_n8n_workflow_with_triggers():
+    """Test workflow validation with triggers array."""
+    valid_workflow_with_triggers = {
+        "name": "Webhook Workflow",
+        "nodes": [
+            {
+                "id": "webhook-1",
+                "name": "Webhook Trigger",
+                "type": "n8n-nodes-base.webhook",
+                "typeVersion": 1,
+                "position": [0, 0]
+            },
+            {
+                "id": "process-1",
+                "name": "Process Data",
+                "type": "n8n-nodes-base.function",
+                "typeVersion": 1,
+                "position": [300, 0]
+            }
+        ],
+        "connections": {},
+        "triggers": [
+            {
+                "nodeId": "webhook-1",
+                "kind": "webhook",
+                "enabled": True,
+                "meta": {
+                    "path": "/api/webhook",
+                    "method": "POST"
+                }
+            }
+        ]
+    }
+    assert N8NWorkflow.validate(valid_workflow_with_triggers)
+    
+    # Empty triggers array is valid
+    workflow_empty_triggers = valid_workflow_with_triggers.copy()
+    workflow_empty_triggers["triggers"] = []
+    assert N8NWorkflow.validate(workflow_empty_triggers)
+    
+    # Workflow without triggers is valid (optional field)
+    workflow_no_triggers = valid_workflow_with_triggers.copy()
+    del workflow_no_triggers["triggers"]
+    assert N8NWorkflow.validate(workflow_no_triggers)
+    
+    # Invalid triggers array (not a list)
+    invalid_workflow = valid_workflow_with_triggers.copy()
+    invalid_workflow["triggers"] = "not a list"
+    assert not N8NWorkflow.validate(invalid_workflow)
+    
+    # Invalid trigger in array
+    invalid_workflow = valid_workflow_with_triggers.copy()
+    invalid_workflow["triggers"] = [{"nodeId": "node-1"}]  # missing kind
+    assert not N8NWorkflow.validate(invalid_workflow)
+    
+    # Multiple triggers
+    workflow_multiple_triggers = valid_workflow_with_triggers.copy()
+    workflow_multiple_triggers["triggers"] = [
+        {"nodeId": "webhook-1", "kind": "webhook"},
+        {"nodeId": "schedule-1", "kind": "schedule"},
+        {"nodeId": "email-1", "kind": "email", "enabled": False}
+    ]
+    assert N8NWorkflow.validate(workflow_multiple_triggers)