app/services/validation.py

import re
from collections import defaultdict
from typing import Protocol

from app.models.workflow import ValidationIssue, ValidationResult, WorkflowDocument
from app.services.catalog import SUPPORTED_NODE_TYPES

EXPRESSION_RE = re.compile(r"^=.*\{\{.+\}\}.*$", re.DOTALL)


class ValidationRule(Protocol):
    code: str

    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]: ...


class TriggerRule:
    code = "missing_trigger"

    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
        if any(node.data.category == "trigger" for node in workflow.nodes):
            return []
        return [
            ValidationIssue(
                code=self.code,
                severity="error",
                message="Workflow does not have a trigger node.",
                suggestion="Add a Webhook, Schedule Trigger, or service trigger.",
            )
        ]


class CredentialRule:
    code = "missing_credentials"

    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
        issues = []
        for node in workflow.nodes:
            for credential in (node.data.credentials or {}).values():
                if not credential.id:
                    issues.append(
                        ValidationIssue(
                            code=self.code,
                            severity="error",
                            nodeId=node.id,
                            message=f"{node.data.label} has an unconfigured credential.",
                            suggestion=f"Select a {credential.type} credential before activation.",
                        )
                    )
        return issues


class ConnectionRule:
    code = "broken_connection"

    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
        ids = {node.id for node in workflow.nodes}
        issues = []
        for edge in workflow.edges:
            if edge.source not in ids or edge.target not in ids:
                issues.append(
                    ValidationIssue(
                        code=self.code,
                        severity="error",
                        message=f"Connection {edge.id} references a missing node.",
                        suggestion="Delete the connection or reconnect it to an existing node.",
                    )
                )
        connected = {edge.source for edge in workflow.edges} | {
            edge.target for edge in workflow.edges
        }
        for node in workflow.nodes:
            if len(workflow.nodes) > 1 and node.id not in connected:
                issues.append(
                    ValidationIssue(
                        code="disconnected_node",
                        severity="warning",
                        nodeId=node.id,
                        message=f"{node.data.label} is not connected.",
                        suggestion="Connect or remove this node.",
                    )
                )
        return issues


class LoopRule:
    code = "infinite_loop"

    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
        graph: dict[str, list[str]] = defaultdict(list)
        for edge in workflow.edges:
            graph[edge.source].append(edge.target)
        visiting: set[str] = set()
        visited: set[str] = set()

        def has_cycle(node_id: str) -> bool:
            if node_id in visiting:
                return True
            if node_id in visited:
                return False
            visiting.add(node_id)
            if any(has_cycle(target) for target in graph[node_id]):
                return True
            visiting.remove(node_id)
            visited.add(node_id)
            return False

        if any(has_cycle(node.id) for node in workflow.nodes if node.id not in visited):
            return [
                ValidationIssue(
                    code=self.code,
                    severity="error",
                    message="Workflow contains a cycle that may execute indefinitely.",
                    suggestion="Break the cycle or add an explicit loop termination condition.",
                )
            ]
        return []


class ParameterRule:
    code = "empty_parameters"

    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
        issues = []
        for node in workflow.nodes:
            if node.data.category != "trigger" and not node.data.parameters:
                issues.append(
                    ValidationIssue(
                        code=self.code,
                        severity="warning",
                        nodeId=node.id,
                        message=f"{node.data.label} has no parameters.",
                        suggestion="Configure the required operation and resource fields.",
                    )
                )
        return issues


class ExpressionRule:
    code = "invalid_expression"

    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
        issues = []

        def inspect(value, node_id: str, label: str) -> None:
            if isinstance(value, dict):
                for nested in value.values():
                    inspect(nested, node_id, label)
            elif isinstance(value, list):
                for nested in value:
                    inspect(nested, node_id, label)
            elif isinstance(value, str) and ("{{" in value or "}}" in value):
                if value.count("{{") != value.count("}}") or not EXPRESSION_RE.match(value):
                    issues.append(
                        ValidationIssue(
                            code=self.code,
                            severity="error",
                            nodeId=node_id,
                            message=f"{label} contains an invalid n8n expression.",
                            suggestion="Expressions should start with = and use balanced {{ }} braces.",
                        )
                    )

        for node in workflow.nodes:
            inspect(node.data.parameters, node.id, node.data.label)
        return issues


class SupportedNodeRule:
    code = "unsupported_node"

    def validate(self, workflow: WorkflowDocument) -> list[ValidationIssue]:
        return [
            ValidationIssue(
                code=self.code,
                severity="warning",
                nodeId=node.id,
                message=f"{node.data.label} is not in the bundled node catalog.",
                suggestion="Confirm that the node package is installed on the target n8n instance.",
            )
            for node in workflow.nodes
            if node.data.type not in SUPPORTED_NODE_TYPES
            and not node.data.type.startswith("@n8n/n8n-nodes-langchain.")
        ]


class Validator:
    def __init__(self) -> None:
        self.rules: list[ValidationRule] = []

    def register(self, rule: ValidationRule) -> None:
        self.rules.append(rule)

    def validate(self, workflow: WorkflowDocument) -> ValidationResult:
        issues = [issue for rule in self.rules for issue in rule.validate(workflow)]
        deductions = {"error": 18, "warning": 6, "info": 1}
        score = max(0, 100 - sum(deductions[issue.severity] for issue in issues))
        return ValidationResult(
            valid=not any(issue.severity == "error" for issue in issues),
            score=score,
            issues=issues,
        )


validator = Validator()
for validation_rule in (
    TriggerRule(),
    CredentialRule(),
    ConnectionRule(),
    LoopRule(),
    ParameterRule(),
    ExpressionRule(),
    SupportedNodeRule(),
):
    validator.register(validation_rule)