Neo4j Orphan Node Audit and Automated Cleanup with Kestra

id: neo4j-orphan-node-cleanup
namespace: company.team
description: Audit Neo4j graph databases for disconnected orphan nodes, support
  dry-run alerts, and execute automated Cypher deletions.

inputs:
  - id: neo4j_url
    type: STRING
    defaults: bolt://neo4j:7687
    displayName: Neo4j Connection URL
    description: Bolt or Neo4j protocol URL for the target database.

  - id: neo4j_username
    type: STRING
    defaults: neo4j
    displayName: Neo4j Username
    description: Username for basic authentication.

  - id: node_label
    type: SELECT
    values:
      - "Order"
      - "Customer"
      - "Product"
    defaults: "Order"
    displayName: Node Label
    description: Target node label to audit for orphaned instances with zero
      relationships. Edit this list to match your graph's labels.

  - id: max_orphans
    type: SELECT
    values:
      - "0"
      - "10"
      - "100"
    defaults: "0"
    displayName: Maximum Orphans Threshold
    description: Maximum acceptable count of orphaned nodes before triggering cleanup.

  - id: dry_run
    type: BOOL
    defaults: true
    displayName: Dry Run Mode
    description: When true, audits and alerts on orphan nodes without executing deletions.

  - id: seed_demo_graph
    type: BOOL
    defaults: true
    displayName: Seed Demo Graph
    description: When true, populates a small demonstration graph with connected and
      orphaned nodes.

triggers:
  - id: weekly_graph_hygiene
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 0 * * 0"
    disabled: true
    inputs:
      seed_demo_graph: false
      dry_run: true

tasks:
  - id: seed_demo_nodes
    type: io.kestra.plugin.core.flow.If
    description: Seed test nodes carrying demo_seed marker when seed_demo_graph is enabled.
    condition: "{{ inputs.seed_demo_graph }}"
    then:
      - id: cleanup_demo_data
        type: io.kestra.plugin.neo4j.Query
        description: Purge previous test nodes carrying the demo_seed marker.
        url: "{{ inputs.neo4j_url }}"
        username: "{{ inputs.neo4j_username }}"
        password: "{{ secret('NEO4J_PASSWORD') }}"
        storeType: NONE
        query: |
          MATCH (n) WHERE n.demo_seed = true DETACH DELETE n

      - id: create_demo_data
        type: io.kestra.plugin.neo4j.Query
        description: Create connected Customer-PLACED->Order nodes alongside isolated
          orphan Order nodes tagged with demo_seed marker.
        url: "{{ inputs.neo4j_url }}"
        username: "{{ inputs.neo4j_username }}"
        password: "{{ secret('NEO4J_PASSWORD') }}"
        storeType: NONE
        query: |
          CREATE (:Customer {id: "cust_101", name: "Alice", demo_seed: true})-[:PLACED]->(:Order {id: "ord_201", status: "COMPLETED", demo_seed: true}), (:Order {id: "ord_991", status: "ORPHANED", demo_seed: true}), (:Order {id: "ord_992", status: "ORPHANED", demo_seed: true})

  - id: count_orphan_nodes
    type: io.kestra.plugin.neo4j.Query
    description: Query Neo4j for nodes matching node_label that have zero incoming
      or outgoing relationships.
    url: "{{ inputs.neo4j_url }}"
    username: "{{ inputs.neo4j_username }}"
    password: "{{ secret('NEO4J_PASSWORD') }}"
    storeType: FETCHONE
    query: |
      MATCH (n:{{ inputs.node_label }})
      WHERE NOT (n)--()
      RETURN count(n) AS orphan_count, collect(n.id)[..5] AS sample_ids

  - id: classify_action
    type: io.kestra.plugin.core.debug.Return
    description: Classify the cleanup action based on orphan count, max_orphans
      threshold, and dry_run mode.
    format: "{{ (outputs.count_orphan_nodes is not defined or
      outputs.count_orphan_nodes.row.orphan_count is not defined or
      outputs.count_orphan_nodes.row.orphan_count <= (inputs.max_orphans | int))
      ? 'CLEAN' : (inputs.dry_run ? 'DRY_RUN' : 'CLEANUP') }}"

  - id: route_action
    type: io.kestra.plugin.core.flow.Switch
    description: Route graph maintenance actions based on classified outcome.
    value: "{{ outputs.classify_action.value }}"
    cases:
      CLEAN:
        - id: log_clean_state
          type: io.kestra.plugin.core.log.Log
          description: Log clean graph status when orphan count is within threshold.
          message: "Neo4j graph is clean. Found {{
            outputs.count_orphan_nodes.row.orphan_count ?? 0 }} orphan nodes
            (threshold: {{ inputs.max_orphans }})."

      DRY_RUN:
        - id: alert_dry_run
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Post a Slack alert detailing detected orphan nodes during dry-run
            mode without deleting data.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "Dry Run Notice: Detected {{ outputs.count_orphan_nodes.row.orphan_count }} orphaned {{ inputs.node_label }} nodes in Neo4j (threshold: {{ inputs.max_orphans }}). Sample IDs: {{ outputs.count_orphan_nodes.row.sample_ids }}. No nodes deleted."
            }

      CLEANUP:
        - id: purge_orphan_nodes
          type: io.kestra.plugin.neo4j.Query
          description: Execute Cypher deletion for orphaned nodes matching the label.
          url: "{{ inputs.neo4j_url }}"
          username: "{{ inputs.neo4j_username }}"
          password: "{{ secret('NEO4J_PASSWORD') }}"
          storeType: NONE
          query: |
            MATCH (n:{{ inputs.node_label }})
            WHERE NOT (n)--()
            DETACH DELETE n

        - id: verify_recount
          type: io.kestra.plugin.neo4j.Query
          description: Re-count remaining orphan nodes post-cleanup for verification.
          url: "{{ inputs.neo4j_url }}"
          username: "{{ inputs.neo4j_username }}"
          password: "{{ secret('NEO4J_PASSWORD') }}"
          storeType: FETCHONE
          query: |
            MATCH (n:{{ inputs.node_label }})
            WHERE NOT (n)--()
            RETURN count(n) AS remaining_count

        - id: alert_cleanup_complete
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Send a Slack alert reporting before and after orphan node counts.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "Cleanup Complete: Purged orphaned {{ inputs.node_label }} nodes in Neo4j. Before count: {{ outputs.count_orphan_nodes.row.orphan_count }}. Remaining count: {{ outputs.verify_recount.row.remaining_count }}."
            }

    defaults:
      - id: log_default_route
        type: io.kestra.plugin.core.log.Log
        description: Fallback log for unclassified routing outcomes.
        message: "Neo4j maintenance execution completed with outcome: {{
          outputs.classify_action.value }}"

errors:
  - id: notify_flow_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post a Slack alert if an unexpected flow-level execution error occurs.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Alert: Neo4j orphan node cleanup workflow failed unexpectedly. Execution ID: {{ execution.id }}."
      }

Automate Neo4j graph database maintenance and hygiene directly inside Kestra. Over time, high-throughput graph databases accumulate disconnected orphan nodes during microservice updates or partial bulk ingestion runs. This blueprint audits target node labels for zero-relationship instances, supports safe dry-run reporting, executes automated Cypher deletions when enabled, and posts structured before-and-after metrics to Slack.

How it works

A scheduled trigger weekly_graph_hygiene (io.kestra.plugin.core.trigger.Schedule, disabled by default) executes the workflow on a weekly cadence with dry_run: true.
Optional task seed_demo_nodes (io.kestra.plugin.core.flow.If) populates a test dataset of connected Customer-PLACED->Order nodes and isolated Order nodes when seed_demo_graph: true. The seed process runs cleanup_demo_data (MATCH (n) WHERE n.demo_seed = true DETACH DELETE n) followed by create_demo_data in separate tasks, strictly scoped to nodes carrying demo_seed: true.
count_orphan_nodes (io.kestra.plugin.neo4j.Query) executes Cypher matching nodes of label node_label where WHERE NOT (n)--() and returns total orphan_count and a list of sample_ids. Node labels are bounded by a predefined SELECT dropdown list.
classify_action (io.kestra.plugin.core.debug.Return) evaluates orphan_count, max_orphans, and dry_run into one of three outcome states: CLEAN, DRY_RUN, or CLEANUP.
route_action (io.kestra.plugin.core.flow.Switch) dispatches workflow execution based on the classified outcome:
- On CLEAN, log_clean_state (io.kestra.plugin.core.log.Log) logs a clean status message.
- On DRY_RUN, alert_dry_run (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) dispatches a Slack alert detailing detected orphan counts without mutating database state.
- On CLEANUP, purge_orphan_nodes (io.kestra.plugin.neo4j.Query) executes DETACH DELETE n on orphaned nodes, verify_recount re-audits the database, and alert_cleanup_complete posts before-and-after metrics to Slack.
A flow-level errors block dispatches a Slack alert if an unexpected infrastructure error occurs.

What you get

Automated graph database hygiene and orphan node detection for Neo4j instances.
Safe default execution: when dry_run: true, no database records are ever modified or deleted.
Bounded input selection restricting target node labels to predefined dropdown options.
Modular outcome classification (CLEAN, DRY_RUN, CLEANUP) using Switch routing.
Slack notification reporting orphan counts, sample IDs, and cleanup verification metrics.

Who it's for

Database administrators and platform engineers maintaining Neo4j production clusters.
Data engineers managing knowledge graph pipelines and graph ETL workloads.
Teams seeking automated graph database maintenance co-located with workflow orchestration.

Why orchestrate this with Kestra

Ad hoc database scripts lack central observability, audit trails, and automated Slack notification loops. Kestra elevates Cypher database maintenance into observable workflows featuring visual topology execution graphs, durable secret management, scheduled execution, and retry capabilities across all environments.

Prerequisites

A running Neo4j 5.x database instance reachable from Kestra workers.
Start a local Neo4j 5.x container with the following command: docker run -d --name neo4j -p 7474:7474 -p 7687:7687 -e NEO4J_AUTH=neo4j/ neo4j:5
If running Kestra via Docker Compose, connect the Neo4j container to Kestra's Docker network: docker network connect YOUR_KESTRA_NETWORK neo4j (Check your exact network name with: docker network ls)
Set neo4j_url to point to the Neo4j container (e.g. bolt://neo4j:7687).

Secrets

NEO4J_PASSWORD: Password for Neo4j database authentication.
SLACK_WEBHOOK_URL: Slack Incoming Webhook URL for alerting.

Note for Open Source Edition (OSS): Secrets can be set as base64-encoded environment variables on the Kestra worker: export SECRET_NEO4J_PASSWORD=$(echo -n "" | base64) export SECRET_SLACK_WEBHOOK_URL=$(echo -n "" | base64)

Inputs

neo4j_url (STRING, default bolt://neo4j:7687): Connection URL for the target Neo4j instance.
neo4j_username (STRING, default neo4j): Basic auth username.
node_label (SELECT, default "Order", options: "Order", "Customer", "Product"): Target node label to audit for orphaned instances.
max_orphans (SELECT, default "0", options: "0", "10", "100"): Orphan threshold before triggering cleanup.
dry_run (BOOLEAN, default true): When true, audits and alerts without executing deletions.
seed_demo_graph (BOOLEAN, default true): Populates a test dataset of connected and orphaned nodes.

Outputs

{{ outputs.count_orphan_nodes.row.orphan_count }}: Initial count of detected orphan nodes.
{{ outputs.count_orphan_nodes.row.sample_ids }}: Sample array of detected orphan node identifiers.
{{ outputs.classify_action.value }}: Classified maintenance outcome (CLEAN, DRY_RUN, or CLEANUP).
{{ outputs.verify_recount.row.remaining_count }}: Verified count of remaining orphan nodes after cleanup.

Quick start

Configure NEO4J_PASSWORD and SLACK_WEBHOOK_URL in Kestra secrets.
Import this blueprint into your namespace (e.g. company.team).
Run the flow with dry_run: true and seed_demo_graph: true to inspect the dry-run notification.
Run the flow with dry_run: false to execute the Cypher deletion and verify node cleanup.
Enable the weekly_graph_hygiene trigger for automated recurring maintenance.

How to extend

Add Cypher constraints and index verification tasks prior to running node purges.
Store historical orphan counts into PostgreSQL or ClickHouse for long-term database SLA reporting.
Expand node label filtering by editing the node_label SELECT values list in YAML.

Pitfalls

Node deletions are permanent; always execute runs with dry_run: true to inspect sample IDs before purging.
An orphan node is defined as having zero relationships; ensure target labels represent entities that must have relationships.
Demonstrations seed data only touches nodes carrying demo_seed: true, but seeding should still only be enabled on test environments.
Edit the node_label SELECT values list in YAML to match your database schema labels.
The demonstration graph dataset is synthetic and intended for initial testing.