feat(backup-monitor): add n8n workflow JSON validator

scripts/validate-n8n-workflow.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+"""Validate an n8n workflow JSON file for structural correctness.
+
+Usage: validate-n8n-workflow.py path/to/workflow.json
+Exit codes: 0 = valid, 1 = invalid, 2 = usage error.
+"""
+import json
+import sys
+from pathlib import Path
+
+KNOWN_MIN_TYPE_VERSIONS = {
+    "n8n-nodes-base.code": 2,
+    "n8n-nodes-base.httpRequest": 4.2,
+    "n8n-nodes-base.webhook": 2,
+    "n8n-nodes-base.set": 3.4,
+    "n8n-nodes-base.scheduleTrigger": 1.2,
+}
+
+REQUIRED_TOP  = ["name", "nodes", "connections"]
+REQUIRED_NODE = ["id", "name", "type", "typeVersion", "position", "parameters"]
+
+
+def validate(workflow):
+    errors, warnings = [], []
+    if not isinstance(workflow, dict):
+        return ["Root must be a JSON object."], []
+    for key in REQUIRED_TOP:
+        if key not in workflow:
+            errors.append(f"Top-level: missing required field '{key}'.")
+    if errors:
+        return errors, warnings
+    if not isinstance(workflow["nodes"], list):
+        errors.append("Top-level 'nodes' must be an array.")
+    if not isinstance(workflow["connections"], dict):
+        errors.append("Top-level 'connections' must be an object.")
+    if errors:
+        return errors, warnings
+
+    seen_names, seen_ids = set(), set()
+    for i, node in enumerate(workflow["nodes"]):
+        loc = f"nodes[{i}]"
+        if not isinstance(node, dict):
+            errors.append(f"{loc}: must be an object.")
+            continue
+        for key in REQUIRED_NODE:
+            if key not in node:
+                errors.append(f"{loc}: missing required field '{key}'.")
+        name = node.get("name")
+        if name:
+            loc = f"nodes[{i}] '{name}'"
+            if name in seen_names:
+                errors.append(f"{loc}: duplicate node name.")
+            seen_names.add(name)
+        nid = node.get("id")
+        if nid:
+            if nid in seen_ids:
+                errors.append(f"{loc}: duplicate node id '{nid}'.")
+            seen_ids.add(nid)
+        pos = node.get("position")
+        if pos is not None and not (
+            isinstance(pos, list) and len(pos) == 2
+            and all(isinstance(v, (int, float)) for v in pos)
+        ):
+            errors.append(f"{loc}: 'position' must be [number, number].")
+        if node.get("parameters") is not None and not isinstance(node["parameters"], dict):
+            errors.append(f"{loc}: 'parameters' must be an object.")
+        creds = node.get("credentials")
+        if isinstance(creds, dict):
+            for ctype, cred in creds.items():
+                if not isinstance(cred, dict) or "name" not in cred:
+                    errors.append(f"{loc}: credentials['{ctype}'] must have a 'name' field.")
+        ntype, tv = node.get("type"), node.get("typeVersion")
+        if isinstance(tv, (int, float)) and ntype in KNOWN_MIN_TYPE_VERSIONS:
+            min_tv = KNOWN_MIN_TYPE_VERSIONS[ntype]
+            if tv < min_tv:
+                warnings.append(f"{loc}: typeVersion {tv} for '{ntype}' is below recommended {min_tv}.")
+
+    for src, outputs in workflow["connections"].items():
+        if src not in seen_names:
+            errors.append(f"connections: source '{src}' does not match any node name.")
+        if not isinstance(outputs, dict):
+            errors.append(f"connections['{src}']: must be an object.")
+            continue
+        for out_type, out_list in outputs.items():
+            if not isinstance(out_list, list):
+                continue
+            for oi, branch in enumerate(out_list):
+                if branch is None:
+                    continue
+                if not isinstance(branch, list):
+                    errors.append(f"connections['{src}']['{out_type}'][{oi}]: must be an array.")
+                    continue
+                for ti, target in enumerate(branch):
+                    if not isinstance(target, dict):
+                        errors.append(f"connections['{src}']['{out_type}'][{oi}][{ti}]: must be an object.")
+                        continue
+                    for k in ("node", "type", "index"):
+                        if k not in target:
+                            errors.append(f"connections['{src}']['{out_type}'][{oi}][{ti}]: missing '{k}'.")
+                    tnode = target.get("node")
+                    if tnode and tnode not in seen_names:
+                        errors.append(f"connections['{src}']['{out_type}'][{oi}][{ti}]: target '{tnode}' not found.")
+    return errors, warnings
+
+
+def main():
+    if len(sys.argv) != 2:
+        print("Usage: validate-n8n-workflow.py <workflow.json>", file=sys.stderr)
+        sys.exit(2)
+    path = Path(sys.argv[1])
+    try:
+        workflow = json.loads(path.read_text())
+    except FileNotFoundError:
+        print(f"File not found: {path}", file=sys.stderr)
+        sys.exit(2)
+    except json.JSONDecodeError as e:
+        print(f"Invalid JSON: {e}", file=sys.stderr)
+        sys.exit(1)
+    errors, warnings = validate(workflow)
+    for w in warnings:
+        print(f"WARN: {w}")
+    for e in errors:
+        print(f"ERROR: {e}", file=sys.stderr)
+    if errors:
+        print(f"\n{len(errors)} error(s), {len(warnings)} warning(s).", file=sys.stderr)
+        sys.exit(1)
+    print(f"OK — {len(workflow['nodes'])} nodes, {len(workflow['connections'])} connections, {len(warnings)} warning(s).")
+
+
+if __name__ == "__main__":
+    main()