Neo4j DAG Cycle Breach and Hierarchy Integrity Gate

id: neo4j-dag-cycle-and-hierarchy-integrity-gate
namespace: company.data

description: |
  Detects directed cycle breaches in Neo4j hierarchy graphs via variable-length Cypher traversal,
  persists a JSON audit certificate of every detected cycle, and conditionally severs the culprit
  closing back-edges to restore the Directed Acyclic Graph invariant.

inputs:
  - id: neo4j_url
    type: STRING
    displayName: Neo4j Bolt URL
    defaults: "bolt://localhost:7687"
    description: Bolt protocol endpoint for the target Neo4j instance.

  - id: target_label
    type: SELECT
    displayName: Target Hierarchy Node Label
    defaults: "OrgUnit"
    values:
      - "OrgUnit"
      - "Role"
      - "Component"
      - "Entity"
      - "Group"
      - "Task"
    description: Node label representing entities within the strict hierarchy or
      dependency graph.

  - id: relationship_type
    type: SELECT
    displayName: Hierarchy Relationship Type
    defaults: "INHERITS"
    values:
      - "INHERITS"
      - "PARENT_OF"
      - "DEPENDS_ON"
      - "PART_OF"
      - "REPORTS_TO"
    description: Directed relationship type that must satisfy the Directed Acyclic
      Graph (DAG) invariant.

  - id: max_search_depth
    type: INT
    displayName: Maximum Cycle Search Depth
    defaults: 8
    description: Upper bound of hops evaluated for cyclic traversals. Prevents
      runaway path expansion on deep trees.

  - id: max_cycles_to_report
    type: INT
    displayName: Maximum Cycles to Report
    defaults: 25
    description: Caps the number of distinct cyclic loops captured in the audit
      certificate per execution.

  - id: auto_sever_cycle
    type: SELECT
    displayName: Auto Sever Culprit Closing Edges
    defaults: "false"
    values:
      - "false"
      - "true"
    description: When true and dry_run is false, automatically deletes the closing
      back-edge relationship to break detected cycles.

  - id: dry_run
    type: SELECT
    displayName: Dry Run Execution Mode
    defaults: "true"
    values:
      - "true"
      - "false"
    description: Must be explicitly set to false along with auto_sever_cycle true
      for any relationship deletion to execute.

triggers:
  - id: daily_hierarchy_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 4 * * *"
    disabled: true

tasks:
  - id: detect_cycles
    type: io.kestra.plugin.neo4j.Query
    description: >-
      Executes variable-length Cypher traversal to detect closed directed
      cycles. Uses elementId filtering on internal path nodes to ensure each
      distinct cycle is captured exactly once anchored at its minimum node
      identifier.
    url: "{{ inputs.neo4j_url }}"
    username: "{{ secret('NEO4J_USERNAME') }}"
    password: "{{ secret('NEO4J_PASSWORD') }}"
    storeType: FETCH
    query: |
      MATCH path = (n:{{ inputs.target_label }})-[r:{{ inputs.relationship_type }}*1..{{ inputs.max_search_depth }}]->(n)
      WHERE all(m IN nodes(path)[1..-1] WHERE elementId(m) > elementId(n))
      WITH n, path, length(path) AS cycle_length,
           [node IN nodes(path) | coalesce(node.name, node.id, elementId(node))] AS path_nodes,
           [rel IN relationships(path) | elementId(rel)] AS path_rel_ids
      RETURN elementId(n) AS anchor_node_id,
             cycle_length,
             path_nodes,
             path_rel_ids,
             last(path_rel_ids) AS culprit_closing_edge_id
      ORDER BY cycle_length ASC, anchor_node_id ASC
      LIMIT {{ inputs.max_cycles_to_report }}

  - id: evaluate_integrity
    type: io.kestra.plugin.core.output.OutputValues
    description: Computes hierarchy integrity status, cycle metrics, and extracts
      closing back-edge identifiers for targeted remediation.
    values:
      cycles_found: "{{ outputs.detect_cycles.size ?? 0 }}"
      status: "{{ (outputs.detect_cycles.size ?? 0) > 0 ? 'CYCLE_BREACH' : 'DAG_VALID'
        }}"
      culprit_edge_ids: "{{ (outputs.detect_cycles.size ?? 0) > 0 ?
        (outputs.detect_cycles.rows | jq('[.[].culprit_closing_edge_id]') |
        first) : [] }}"
      worst_cycle_length: "{{ (outputs.detect_cycles.size ?? 0) > 0 ?
        (outputs.detect_cycles.rows | jq('min_by(.cycle_length) |
        .cycle_length') | first) : 0 }}"

  - id: route_integrity_verdict
    type: io.kestra.plugin.core.flow.Switch
    description: Routes execution based on whether directed cycles breach the DAG invariant.
    value: "{{ outputs.evaluate_integrity.values.status }}"
    cases:
      CYCLE_BREACH:
        - id: generate_audit_certificate
          type: io.kestra.plugin.core.storage.Write
          description: Writes a machine-readable JSON hierarchy audit certificate
            containing the detected cycle traces to Kestra internal storage.
          extension: .json
          content: |
            {
              "audit_timestamp": "{{ execution.startDate }}",
              "target_label": "{{ inputs.target_label }}",
              "relationship_type": "{{ inputs.relationship_type }}",
              "cycles_detected": {{ outputs.evaluate_integrity.values.cycles_found }},
              "shortest_cycle_length": {{ outputs.evaluate_integrity.values.worst_cycle_length }},
              "culprit_closing_edge_ids": {{ outputs.evaluate_integrity.values.culprit_edge_ids | toJson }},
              "cycle_traces": {{ outputs.detect_cycles.rows | toJson }}
            }

        - id: check_remediation_gate
          type: io.kestra.plugin.core.flow.If
          description: Safely severs culprit closing back-edges when auto_sever_cycle is
            enabled and dry_run is disabled.
          condition: "{{ inputs.auto_sever_cycle == 'true' and inputs.dry_run == 'false'
            }}"
          then:
            - id: sever_culprit_closing_edges
              type: io.kestra.plugin.neo4j.Query
              description: >-
                Deletes only the culprit closing relationships identified in the
                audit certificate, breaking every detected directed cycle while
                strictly preserving all node entities and non-cyclical hierarchy
                branches.
              url: "{{ inputs.neo4j_url }}"
              username: "{{ secret('NEO4J_USERNAME') }}"
              password: "{{ secret('NEO4J_PASSWORD') }}"
              storeType: FETCHONE
              query: |
                UNWIND {{ outputs.evaluate_integrity.values.culprit_edge_ids | toJson }} AS closing_rel_id
                MATCH ()-[r:{{ inputs.relationship_type }}]->()
                WHERE elementId(r) = closing_rel_id
                DELETE r
                RETURN count(closing_rel_id) AS severed_count

            - id: verify_dag_restored
              type: io.kestra.plugin.neo4j.Query
              description: Re-runs the cycle detection check to verify that directed cycles
                have been eliminated.
              url: "{{ inputs.neo4j_url }}"
              username: "{{ secret('NEO4J_USERNAME') }}"
              password: "{{ secret('NEO4J_PASSWORD') }}"
              storeType: FETCHONE
              query: |
                MATCH path = (n:{{ inputs.target_label }})-[r:{{ inputs.relationship_type }}*1..{{ inputs.max_search_depth }}]->(n)
                RETURN count(path) AS remaining_cycles

            - id: log_remediation_audit
              type: io.kestra.plugin.core.log.Log
              message: >-
                Remediation applied to Neo4j {{ inputs.target_label }}: severed
                {{ outputs.sever_culprit_closing_edges.row.severed_count }}
                closing edge(s). Remaining cycles: {{
                outputs.verify_dag_restored.row.remaining_cycles }}. Audit
                certificate: {{ outputs.generate_audit_certificate.uri }}.

        - id: alert_cycle_breach_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Dispatches an incident alert to Slack with cycle path lengths,
            affected node IDs, and audit certificate URI.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {
              "text": "🚨 *Neo4j Hierarchy DAG Cycle Breach Detected*\n\n" ~
                      "• *Target Label*: `" ~ inputs.target_label ~ "`\n" ~
                      "• *Relationship*: `[:" ~ inputs.relationship_type ~ "]`\n" ~
                      "• *Cycles Detected*: " ~ outputs.evaluate_integrity.values.cycles_found ~ "\n" ~
                      "• *Shortest Cycle Length*: " ~ outputs.evaluate_integrity.values.worst_cycle_length ~ " hops\n" ~
                      "• *Remediation Mode*: " ~ (inputs.auto_sever_cycle == 'true' and inputs.dry_run == 'false' ? 'AUTOMATICALLY SEVERED' : 'DRY RUN / MANUAL INTERVENTION REQUIRED') ~ "\n" ~
                      "• *Audit Certificate*: `" ~ outputs.generate_audit_certificate.uri ~ "`\n\n" ~
                      "Recursive hierarchy queries may freeze or exhaust memory until resolved."
            } | toJson }}

      DAG_VALID:
        - id: log_dag_healthy
          type: io.kestra.plugin.core.log.Log
          message: >-
            Neo4j hierarchy integrity verified: 0 directed cycles detected for
            label {{ inputs.target_label }} along [:{{ inputs.relationship_type
            }}] up to depth {{ inputs.max_search_depth }}. Directed Acyclic
            Graph invariant intact.

    defaults:
      - id: log_unexpected_state
        type: io.kestra.plugin.core.log.Log
        message: "Unexpected hierarchy evaluation status: {{
          outputs.evaluate_integrity.values.status }}"

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts the data engineering team if the Neo4j query or credentials
      fail during execution.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {
        "text": "❌ *Neo4j Hierarchy Integrity Gate Failed*\n\n" ~
                "Execution `" ~ execution.id ~ "` failed while auditing `" ~ inputs.target_label ~ "` hierarchies on `" ~ inputs.neo4j_url ~ "`. Check Kestra execution logs for driver connection or query errors."
      } | toJson }}