Neo4j Duplicate Relationship Cleanup Guard with Kestra

id: neo4j-duplicate-relationship-cleanup-guard
namespace: company.team

description: |
  Find node pairs in Neo4j that are joined by more than one relationship of the same
  type, report them to Slack and after approval delete the surplus edges so exactly one
  survives, then re-run the same detection and fail the run if any pair is still
  duplicated. No Neo4j constraint caps a node pair at one relationship of a type, so a
  CREATE where a MERGE was meant or a retried load silently leaves parallel identical
  edges behind.
  Those inflate every degree count and double-count weights in path and recommendation
  queries.

inputs:
  - id: url
    type: STRING
    displayName: Neo4j Bolt URL
    defaults: "bolt://localhost:7687"

  - id: user
    type: STRING
    displayName: Neo4j user
    defaults: "neo4j"

  - id: password
    type: STRING
    displayName: Neo4j password
    description: Defaults to the NEO4J_PASSWORD secret. Override only for local testing.
    defaults: "{{ secret('NEO4J_PASSWORD') }}"

  - id: relationship_types
    type: STRING
    displayName: Relationship types to check (comma separated)
    description: >-
      Only these types are grouped and deduplicated. Duplicates are always
      counted per type, so two edges of different types between the same pair
      are never duplicates.
    defaults: "FOLLOWS"

  - id: property_match_mode
    type: SELECT
    displayName: When two edges count as duplicates
    description: >-
      ignore_properties treats every relationship of the same type between the
      same pair as a duplicate, whatever its properties. match_all_properties
      only groups edges whose full property map is identical, so edges that
      differ in any property are kept.
    values:
      - ignore_properties
      - match_all_properties
    defaults: ignore_properties

  - id: survivor_timestamp_property
    type: STRING
    displayName: Timestamp property that picks the survivor
    description: >-
      The edge with the lowest value of this property survives, so the oldest
      edge is kept. Edges missing the property sort last. Ties and an empty
      value here fall back to the lowest elementId, which keeps the choice
      deterministic.
    defaults: "created_at"

  - id: max_pairs
    type: INT
    displayName: Maximum duplicated pairs handled per run
    description: Caps the size of one cleanup transaction. Pairs beyond the cap are
      reported and left for the next run.
    defaults: 50

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for any DELETE to run.
    defaults: true

tasks:
  - id: audit
    type: io.kestra.plugin.neo4j.Query
    description: >-
      Read-only. Groups every relationship of the selected types by start node,
      end node and type (plus the full property map in match_all_properties
      mode) and keeps the groups holding more than one edge. Returns the pair,
      the type, how many edges exist and how many are surplus.
    url: "{{ inputs.url }}"
    username: "{{ inputs.user }}"
    password: "{{ inputs.password }}"
    storeType: FETCH
    query: |
      MATCH (a)-[r]->(b)
      WHERE type(r) IN [t IN split('{{ inputs.relationship_types | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',') | trim(t)]
      WITH a, b, type(r) AS rel_type,
           CASE WHEN '{{ inputs.property_match_mode }}' = 'match_all_properties' THEN properties(r) ELSE null END AS prop_key,
           collect(r) AS rs
      WHERE size(rs) > 1
      RETURN elementId(a) AS start_id,
             elementId(b) AS end_id,
             rel_type,
             size(rs) AS edge_count,
             size(rs) - 1 AS surplus,
             labels(a) AS start_labels,
             labels(b) AS end_labels,
             coalesce(head([n IN [a.name, a.title, a.id] WHERE n IS NOT NULL]), '') AS start_hint,
             coalesce(head([n IN [b.name, b.title, b.id] WHERE n IS NOT NULL]), '') AS end_hint
      ORDER BY surplus DESC, rel_type, start_id, end_id
      LIMIT {{ inputs.max_pairs }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Neo4j duplicate relationship audit on {{ inputs.url }} for types [{{
      inputs.relationship_types }}], mode {{ inputs.property_match_mode }}: {{
      outputs.audit.size }} duplicated group(s). {% for r in outputs.audit.rows
      %}{{ r.rel_type }} {{ r.start_hint }} ({{ r.start_id }}) to {{ r.end_hint
      }} ({{ r.end_id }}): {{ r.edge_count }} edges, {{ r.surplus }} surplus. {%
      endfor %}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit.size > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Neo4j ' ~ inputs.url ~ ': ' ~ outputs.audit.size ~ ' node pair(s) carry duplicate relationships of types [' ~ inputs.relationship_types ~ ']. Surplus edges: ' ~ (outputs.audit.rows | jq('[.[].surplus] | add | tostring') | first) ~ '. Worst: ' ~ (outputs.audit.rows | jq('.[0] | .rel_type + " " + .start_id + " -> " + .end_id + " (" + (.edge_count|tostring) + " edges)"') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Deleting the surplus after approval.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approve
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              Deleting a relationship is not reversible, so a human approves
              before any edge is removed. Resume with approved true to delete
              the surplus, false to leave the graph exactly as it is.
            onResume:
              - id: approved
                type: BOOL
                defaults: false

          - id: apply_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approve.onResume.approved ?? false }}"
            then:
              - id: dedupe
                type: io.kestra.plugin.neo4j.Query
                description: >-
                  Re-groups the duplicates in the same statement that deletes
                  them, so the survivor is chosen against the graph as it is at
                  delete time rather than a stale snapshot. Each group is
                  ordered by the survivor timestamp property then by elementId,
                  the head is kept and tail() is deleted, so exactly one edge
                  survives per group.
                url: "{{ inputs.url }}"
                username: "{{ inputs.user }}"
                password: "{{ inputs.password }}"
                storeType: FETCH
                query: |
                  MATCH (a)-[r]->(b)
                  WHERE type(r) IN [t IN split('{{ inputs.relationship_types | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',') | trim(t)]
                  WITH a, b, type(r) AS rel_type,
                       CASE WHEN '{{ inputs.property_match_mode }}' = 'match_all_properties' THEN properties(r) ELSE null END AS prop_key,
                       collect(r) AS rs
                  WHERE size(rs) > 1
                  WITH a, b, rel_type, prop_key, rs
                  ORDER BY size(rs) DESC, rel_type, elementId(a), elementId(b)
                  LIMIT {{ inputs.max_pairs }}
                  UNWIND rs AS dup
                  WITH a, b, rel_type, prop_key, dup
                  ORDER BY (dup['{{ inputs.survivor_timestamp_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}'] IS NULL), dup['{{ inputs.survivor_timestamp_property | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}'], elementId(dup)
                  WITH a, b, rel_type, prop_key, collect(dup) AS ordered
                  UNWIND tail(ordered) AS surplus_rel
                  WITH elementId(surplus_rel) AS dead_id, surplus_rel
                  DELETE surplus_rel
                  RETURN count(dead_id) AS deleted_count, collect(dead_id)[..25] AS deleted_ids

              - id: reaudit
                type: io.kestra.plugin.neo4j.Query
                description: Read-only. Runs the identical detection query again so the verify
                  step reads the graph rather than trusting the delete count.
                url: "{{ inputs.url }}"
                username: "{{ inputs.user }}"
                password: "{{ inputs.password }}"
                storeType: FETCH
                query: |
                  MATCH (a)-[r]->(b)
                  WHERE type(r) IN [t IN split('{{ inputs.relationship_types | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',') | trim(t)]
                  WITH a, b, type(r) AS rel_type,
                       CASE WHEN '{{ inputs.property_match_mode }}' = 'match_all_properties' THEN properties(r) ELSE null END AS prop_key,
                       collect(r) AS rs
                  WHERE size(rs) > 1
                  RETURN elementId(a) AS start_id,
                         elementId(b) AS end_id,
                         rel_type,
                         size(rs) AS edge_count,
                         size(rs) - 1 AS surplus
                  ORDER BY surplus DESC, rel_type, start_id, end_id
                  LIMIT {{ inputs.max_pairs }}

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                description: Fail if any pair the audit reported is still duplicated after the
                  delete.
                condition: >-
                  {{ {'before': outputs.audit.rows, 'after':
                  (outputs.reaudit.rows ?? [])} | jq('([.before[] | .start_id +
                  "|" + .end_id + "|" + .rel_type] | unique) as $t | [.after[] |
                  .start_id + "|" + .end_id + "|" + .rel_type] | unique | [.[] |
                  select(. as $x | ($t | index($x)) != null)] | length') | first
                  > 0 }}
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: "Pairs still duplicated after the delete: {{ {'before':
                      outputs.audit.rows, 'after': outputs.reaudit.rows} |
                      jq('([.before[] | .start_id + \"|\" + .end_id + \"|\" +
                      .rel_type] | unique) as $t | [.after[] | .start_id + \"|\"
                      + .end_id + \"|\" + .rel_type] | unique | [.[] | select(.
                      as $x | ($t | index($x)) != null)] | join(\", \")') |
                      first }}. A concurrent writer can re-create an edge during
                      the approval window. A user without DELETE on the
                      relationship type cannot remove one. Check the graph and
                      the writer that loads these edges."
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Deduplicated {{ outputs.audit.size }} group(s) on {{
                      inputs.url }}. Edges before: {{ outputs.audit.rows |
                      jq('[.[].edge_count] | add | tostring') | first }}.
                      Surplus deleted: {{ outputs.dedupe.rows[0].deleted_count
                      }}. Groups still duplicated anywhere in scope: {{
                      outputs.reaudit.size }}. {% for r in outputs.audit.rows
                      %}{{ r.rel_type }} {{ r.start_id }} to {{ r.end_id }}: {{
                      r.edge_count }} edges before, 1 after. {% endfor %}
            else:
              - id: approval_denied
                type: io.kestra.plugin.core.log.Log
                message: "Approval not granted, no relationship was deleted. The audit list
                  stands."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No node pair in {{ inputs.url }} carries duplicate relationships of
          types [{{ inputs.relationship_types }}] under mode {{
          inputs.property_match_mode }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Neo4j duplicate relationship guard failed on ' ~ inputs.url ~ ' (execution ' ~ execution.id ~ '). Duplicate state unknown, check the logs.'} | toJson }}

triggers:
  - id: weekly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, always in safe mode.
    cron: "0 4 * * 1"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

Neo4j constraints can make a property unique. None can cap a node pair at one relationship of a type. Nothing stops two identical (a)-[:FOLLOWS]->(b) edges existing side by side, so one CREATE where a MERGE was meant, one retried load or one import replayed after a partial failure leaves parallel edges behind and the database accepts all of them.

The damage is quiet. Every degree calculation is inflated. Every count(r) is wrong. Weighted path and recommendation queries count the same edge two or three times, so the result is not just slower, it is different. Nobody gets an error, so the drift sits in the graph until someone compares a number against the source system.

This blueprint groups the relationships of the types you name by start node, end node and type, keeps the groups that hold more than one edge and reports them to Slack. When you turn remediation on and a human approves, it deletes the surplus edges so exactly one survives per group, runs the identical detection again and fails the run if any pair it reported is still duplicated.

This blueprint was created by zkasuran.

Safety

Deleting a relationship cannot be undone, so three things sit between a finding and any delete: auto_remediate must be true, dry_run must be false and a human must resume a Pause with approved: true. The default run only reports.

The survivor is deterministic, never arbitrary. Each group is ordered by the property named in survivor_timestamp_property (ascending, so the oldest edge wins), then by elementId as a tie-break. Edges missing that property sort last, so a load that forgot to stamp a timestamp loses to one that did not. Two runs over the same graph therefore keep the same edge.

Grouping is always scoped to one type, so a pair joined by :FOLLOWS and :BLOCKS is never treated as duplicated. relationship_types narrows the scan further. max_pairs caps how much one transaction touches.

How it works

audit (io.kestra.plugin.neo4j.Query, read-only, storeType: FETCH) matches every relationship whose type is in relationship_types, groups by start node, end node and type, then keeps the groups where size(collect(r)) > 1. In match_all_properties mode the full property map joins the grouping key, so only edges that are identical property for property group together. It returns the pair, the type, the edge count, the surplus count, both label sets and a readable hint from name, title or id when the nodes carry one.
summary logs every duplicated group it found.
route (io.kestra.plugin.core.flow.If on outputs.audit.size > 0):
- clean: clean logs that no pair is duplicated.
- findings: alert posts the pair count, the total surplus and the worst group to Slack, saying whether it is about to delete.
  - remediate_gate runs only when auto_remediate is true and dry_run is false.
  - approve is a Pause. A human resumes it with approved true to go ahead or false to stop.
  - apply_gate runs the delete only when approved is true.
  - dedupe (io.kestra.plugin.neo4j.Query) re-groups and deletes in one statement, so the survivor is chosen against the graph at delete time. It orders each group by the survivor property then elementId, keeps the head and deletes tail(), then returns the deleted count with a sample of deleted element ids.
  - reaudit runs the identical detection query again.
  - verify_result fails the run if any pair the audit reported still comes back duplicated.
errors alerts Slack when the flow itself fails, for example an unreachable instance or a user without delete rights, so a failed audit is never read as "nothing wrong".

What you get

A list of the node pairs that are actually over-connected, straight from Cypher, with the edge count and the surplus per pair.
A choice about what counts as a duplicate: any edge of the same type or only edges identical in every property.
A deterministic survivor, documented and repeatable, instead of "whichever one Cypher returned first".
Type scoping, so a pair legitimately joined by two different relationship types is never touched.
Three gates between a finding and an irreversible delete.
Proof the cleanup landed: the run re-reads the graph and fails loudly if a pair is still duplicated, which is what happens when a writer re-creates the edge during the approval window.

Who it's for

Teams loading a graph from a relational source or an event stream, where a retry or a CREATE in place of a MERGE leaves parallel edges.
Graph and data engineers whose degree counts, weights or recommendation scores stopped matching the source system.
Platform teams who want duplicate edges found on a schedule and cleaned only with an approval, rather than discovered during an incident.
Anyone evaluating the Neo4j plugin who wants a worked example of grouped detection, a gated write and a re-read verification in one flow.

Why orchestrate this with Kestra

The detection is one Cypher query. Doing something about it safely is a sequence: capture the duplicated pairs, tell the team, wait for a human, delete the surplus with a survivor rule you can defend, read the graph again and fail if the cleanup did not take. Kestra runs that sequence on a schedule behind two inputs and an approval, keeps a history of which pairs were cleaned and when. It alerts Slack on both the finding and on a failure. A cron job with a DELETE in it has no approval step and no proof.

Tested end to end

On Kestra 2.0.5 OSS against Neo4j 5 Community, with a local HTTP sink standing in for Slack. The graph was seeded with ten :User nodes and five pairs: one pair joined by three :FOLLOWS edges, one pair joined by two byte-identical :FOLLOWS edges, one pair joined by two :FOLLOWS edges that differ only in a source property, one single :FOLLOWS edge plus one pair joined by one :FOLLOWS and one :BLOCKS. That is 9 :FOLLOWS edges, 4 of them surplus. cypher-shell counted the relationships per pair before and after every run.

Run	Inputs	Result
1	defaults, graph already clean	0 groups, `clean` logged. No alert
2	defaults, seeded duplicates	3 groups reported (3, 2 and 2 edges), alert sent, nothing deleted
3	`auto_remediate: true`, `dry_run: true`	Same report, delete held. `cypher-shell` still counts 9 `:FOLLOWS`
4	`auto_remediate: true`, `dry_run: false`, approved false	Report, `approve` resumed with false, nothing deleted
5	`auto_remediate: true`, `dry_run: false`, approved true	4 surplus edges deleted, all 3 pairs down to exactly 1, `reaudit` empty, `verify_result` passed. The oldest edge survived in every group. The single edge and the two different-type edges untouched
6	`property_match_mode: match_all_properties`	Only the byte-identical group reported. The pair differing in `source` and the pair differing in `created_at` are left alone
7	unreachable URL	`audit` failed, `errors` alert sent
8	`max_pairs: 3`, a writer adds a bigger duplicated pair during the approval window	The new pair takes a slot in the capped batch, one audited pair is not cleaned, `verify_result` fails the run and names it

Exact execution ids and the before and after relationship counts are in the pull request notes.

Prerequisites

A reachable Neo4j 5.x instance. The flow uses plain Cypher (MATCH, collect, tail, DELETE), so it works on Community and Enterprise.
For the audit, a user that can read the relationships in scope. For remediation, a user allowed to delete them.
A Slack incoming webhook for alerts.
Local testing:
- docker run -d --name neo4j -p 7687:7687 -e NEO4J_AUTH=neo4j/testpassword neo4j:5;
- seed a duplicate from another shell: cypher-shell -u neo4j -p testpassword "CREATE (a:User {name:'ada'}), (b:User {name:'bob'});" then run MATCH (a:User {name:'ada'}),(b:User {name:'bob'}) CREATE (a)-[:FOLLOWS]->(b); twice.

Secrets

NEO4J_PASSWORD: password for the Neo4j user, the default for the password input.
SLACK_WEBHOOK_URL: Slack incoming webhook used by alert and the errors block.
In Kestra OSS, provide each one as a base64-encoded environment variable prefixed SECRET_, for example SECRET_NEO4J_PASSWORD=$(echo -n 'testpassword' | base64).

Inputs

Input	Default	Purpose
`url`	`bolt://localhost:7687`	Bolt endpoint
`user`	`neo4j`	Neo4j user
`password`	`{{ secret('NEO4J_PASSWORD') }}`	Password, from the secret by default
`relationship_types`	`FOLLOWS`	Comma-separated types to group and deduplicate
`property_match_mode`	`ignore_properties`	`ignore_properties` or `match_all_properties`
`survivor_timestamp_property`	`created_at`	Lowest value survives, `elementId` breaks ties
`max_pairs`	`50`	Cap on duplicated groups handled per run
`auto_remediate`	`false`	Allow delete
`dry_run`	`true`	Must be `false` for any delete to run

Quick start

Add the NEO4J_PASSWORD and SLACK_WEBHOOK_URL secrets.
Import the flow and set url and user.
Put the relationship types your ingest writes into relationship_types and run it once with the defaults.
Read the report. Decide whether edges differing only in properties are duplicates for your model and set property_match_mode to match.
Set survivor_timestamp_property to the property your loader stamps, so the oldest edge is the one that survives.
Run it with auto_remediate: true and dry_run: false, approve the Pause, then the surplus edges are deleted and verified.
Enable the weekly trigger, which always runs in safe mode.

How to extend

Keep the newest edge instead of the oldest by reversing the order in dedupe, for example when the latest load carries the correct weight.
Merge properties before deleting, so the survivor inherits the union of the duplicates rather than only its own values.
Narrow the scan with a label filter, MATCH (a:User)-[r]->(b:User), on a graph where only a few labels are loaded non-idempotently.
Treat direction as irrelevant for symmetric types by matching (a)-[r]-(b) and ordering the pair by elementId before grouping.
Follow the cleanup with a MERGE-based loader change, so the duplicates stop arriving instead of being swept weekly.
Report to a ticket instead of deleting, for an advisory deployment on a graph nobody is allowed to change automatically.

Neo4j Duplicate Relationship Cleanup Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Neo4j Duplicate Relationship Cleanup Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

More Data Blueprints

New to Kestra?