MongoDB Dangling Reference Orphan Cleanup Guard with Kestra

id: mongodb-dangling-reference-orphan-cleanup-guard
namespace: company.team

description: |
  Find the MongoDB child documents whose manual reference points at a parent document that no
  longer exists. MongoDB has no foreign keys, so deleting a parent leaves every child pointing at
  a key nothing resolves. Those children keep working until something joins on them: a `$lookup`
  returns an empty array, an inner join drops the row, a report that counts by parent silently
  loses it. This flow joins the child collection to the parent collection server side with
  `$lookup`, treats an empty join result as a dangling reference, spares any child written too
  recently to judge, deletes the aged orphans behind a safe mode gate and an approval, then runs
  the join again and fails the run if an aged orphan survived.

inputs:
  - id: connection_uri
    type: STRING
    displayName: MongoDB connection string
    description: >-
      Leave empty in production so the flow reads the MONGODB_URI secret. Set it
      to a plain connection string only for a local test run.
    defaults: ""

  - id: database
    type: STRING
    displayName: Database
    defaults: appdata

  - id: child_collection
    type: STRING
    displayName: Child collection
    description: The collection that holds the references, for example orders.
    defaults: orders

  - id: parent_collection
    type: STRING
    displayName: Parent collection
    description: The collection the references point at, for example customers.
    defaults: customers

  - id: reference_field
    type: STRING
    displayName: Reference field on the child
    description: >-
      The field on a child document that holds the parent key, for example
      customerId. A child that is missing this field or holds null in it, is
      never read as an orphan.
    defaults: customerId

  - id: parent_key_field
    type: STRING
    displayName: Key field on the parent
    description: >-
      The field on a parent document the reference resolves against. Usually
      _id. The join is a strict BSON comparison, so the two fields must hold the
      same type.
    defaults: _id

  - id: age_field
    type: STRING
    displayName: Child write time field
    description: >-
      Optional BSON date field that records when the child was written. Leave
      empty to take the write time from the timestamp inside a stock ObjectId
      _id. A child whose age cannot be read at all counts as brand new, so it is
      spared.
    defaults: ""

  - id: min_orphan_age_minutes
    type: INT
    displayName: Minimum orphan age in minutes
    description: >-
      A dangling child younger than this is reported but never deleted. It
      exists because a child can legitimately be written microseconds before its
      parent. It can also land in the middle of a multi document insert that is
      still in flight. Raise it if your writers insert children first.
    defaults: 15

  - id: max_orphan_ratio
    type: FLOAT
    displayName: Maximum orphan share
    description: >-
      Refuse to delete when the orphan share of all referencing children is
      above this. A share near 1 almost always means the join itself is wrong,
      usually a type mismatch between the reference field and the parent key,
      not that every row lost its parent.
    defaults: 0.5

  - id: scan_limit
    type: INT
    displayName: Orphans named per report
    description: >-
      Ceiling on how many orphans the report lists. The counts stay exact and
      server side, so the cap only bounds the sample.
    defaults: 25

  - id: delete_limit
    type: INT
    displayName: Orphans deleted per run
    description: >-
      Ceiling on how many aged orphans one run deletes, so a huge backlog cannot
      become one giant write. If more are aged than this, the run clears part of
      the backlog then the re-join fails until a later run finishes the job.
    defaults: 50000

  - id: batch_size
    type: INT
    displayName: Document ids per delete batch
    description: The delete runs in batches of this many ids, one batch at a time.
    defaults: 500

  - id: approval_timeout
    type: STRING
    displayName: Approval timeout
    description: ISO-8601 duration to wait for an answer before the pause expires.
    defaults: "PT2H"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Allow the delete. Still needs dry_run false and an approval.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, before any document is
      deleted.
    defaults: true

variables:
  cutoff_iso: '{{ execution.startDate | date("yyyy-MM-dd''T''HH:mm:ss.SSS''Z''",
    timeZone="UTC") }}'

  type_probe_pipeline: >-
    {{ {'pc': inputs.parent_collection, 'rf': inputs.reference_field, 'pk':
    inputs.parent_key_field} | jq('.pc as $pc | .rf as $rf | .pk as $pk |
    [{"$match": {($rf): {"$exists": true}}}, {"$group": {"_id": {"$type": ("$" +
    $rf)}, "n": {"$sum": 1}}}, {"$project": {"_id": 0, "side": {"$literal":
    "child_reference"}, "bson_type": "$_id", "docs": "$n"}}, {"$unionWith":
    {"coll": $pc, "pipeline": [{"$match": {($pk): {"$exists": true}}},
    {"$group": {"_id": {"$type": ("$" + $pk)}, "n": {"$sum": 1}}}, {"$project":
    {"_id": 0, "side": {"$literal": "parent_key"}, "bson_type": "$_id", "docs":
    "$n"}}]}}]') | first | toJson }}

  type_mismatch: >-
    {{ {'rows': outputs.type_probe.rows} | jq('[.rows[] | select(.side ==
    "child_reference") | .bson_type] as $c | [.rows[] | select(.side ==
    "parent_key") | .bson_type] as $p | (if (($c | length) > 0) and (($p |
    length) > 0) and ((($c - ($c - $p)) | length) == 0) then "true" else "false"
    end)') | first }}

  detect_pipeline: >-
    {{ {'pc': inputs.parent_collection, 'rf': inputs.reference_field, 'pk':
    inputs.parent_key_field, 'af': inputs.age_field, 'ms':
    inputs.min_orphan_age_minutes * 60000, 'sd': render(vars.cutoff_iso), 'sl':
    inputs.scan_limit, 'dl': inputs.delete_limit} | jq('.pc as $pc | .rf as $rf
    | .pk as $pk | .af as $af | .ms as $ms | .sd as $sd | .sl as $sl | .dl as
    $dl | {"$subtract": [{"$date": $sd}, $ms]} as $cut | {"$convert": {"input":
    "$_id", "to": "date", "onError": "$$NOW", "onNull": "$$NOW"}} as $fromid |
    (if $af == "" then $fromid else {"$ifNull": [("$" + $af), $fromid]} end) as
    $born | [{"$match": {($rf): {"$exists": true, "$ne": null}}}, {"$lookup":
    {"from": $pc, "localField": $rf, "foreignField": $pk, "as": "_parent"}},
    {"$set": {"_dangling": {"$eq": [{"$size": "$_parent"}, 0]}, "_born": $born,
    "_cut": $cut}}, {"$set": {"_aged": {"$lte": ["$_born", "$_cut"]}}},
    {"$facet": {"sample": [{"$match": {"_dangling": true}}, {"$sort": {"_born":
    1}}, {"$limit": $sl}, {"$project": {"_id": 0, "doc_id": {"$toString":
    "$_id"}, "reference": {"$toString": ("$" + $rf)}, "written_at": "$_born",
    "past_age_gate": "$_aged"}}], "targets": [{"$match": {"_dangling": true,
    "_aged": true}}, {"$sort": {"_born": 1}}, {"$limit": $dl}, {"$group":
    {"_id": "all", "ids": {"$push": {"$toString": "$_id"}}}}], "totals":
    [{"$group": {"_id": "all", "referencing": {"$sum": 1}, "orphans": {"$sum":
    {"$cond": ["$_dangling", 1, 0]}}, "deletable": {"$sum": {"$cond": [{"$and":
    ["$_dangling", "$_aged"]}, 1, 0]}}, "spared_fresh": {"$sum": {"$cond":
    [{"$and": ["$_dangling", {"$not": ["$_aged"]}]}, 1, 0]}}}}]}}, {"$project":
    {"referencing": {"$ifNull": [{"$arrayElemAt": ["$totals.referencing", 0]},
    0]}, "orphans": {"$ifNull": [{"$arrayElemAt": ["$totals.orphans", 0]}, 0]},
    "deletable": {"$ifNull": [{"$arrayElemAt": ["$totals.deletable", 0]}, 0]},
    "spared_fresh": {"$ifNull": [{"$arrayElemAt": ["$totals.spared_fresh", 0]},
    0]}, "orphan_ids": {"$ifNull": [{"$arrayElemAt": ["$targets.ids", 0]}, []]},
    "sample": "$sample"}}, {"$set": {"targeted": {"$size": "$orphan_ids"},
    "orphan_ratio": {"$cond": [{"$gt": ["$referencing", 0]}, {"$round":
    [{"$divide": ["$orphans", "$referencing"]}, 4]}, 0]}}}]') | first | toJson
    }}

  reverify_pipeline: >-
    {{ {'pc': inputs.parent_collection, 'rf': inputs.reference_field, 'pk':
    inputs.parent_key_field, 'af': inputs.age_field, 'ms':
    inputs.min_orphan_age_minutes * 60000, 'sd': render(vars.cutoff_iso)} |
    jq('.pc as $pc | .rf as $rf | .pk as $pk | .af as $af | .ms as $ms | .sd as
    $sd | {"$subtract": [{"$date": $sd}, $ms]} as $cut | {"$convert": {"input":
    "$_id", "to": "date", "onError": "$$NOW", "onNull": "$$NOW"}} as $fromid |
    (if $af == "" then $fromid else {"$ifNull": [("$" + $af), $fromid]} end) as
    $born | [{"$match": {($rf): {"$exists": true, "$ne": null}}}, {"$lookup":
    {"from": $pc, "localField": $rf, "foreignField": $pk, "as": "_parent"}},
    {"$set": {"_dangling": {"$eq": [{"$size": "$_parent"}, 0]}, "_born": $born,
    "_cut": $cut}}, {"$set": {"_aged": {"$lte": ["$_born", "$_cut"]}}},
    {"$facet": {"totals": [{"$group": {"_id": "all", "referencing": {"$sum": 1},
    "orphans": {"$sum": {"$cond": ["$_dangling", 1, 0]}}, "deletable": {"$sum":
    {"$cond": [{"$and": ["$_dangling", "$_aged"]}, 1, 0]}}}}], "worst":
    [{"$match": {"_dangling": true, "_aged": true}}, {"$sort": {"_born": 1}},
    {"$limit": 5}, {"$project": {"_id": 0, "doc_id": {"$toString": "$_id"},
    "reference": {"$toString": ("$" + $rf)}}}]}}, {"$project":
    {"still_referencing": {"$ifNull": [{"$arrayElemAt": ["$totals.referencing",
    0]}, 0]}, "remaining_orphans": {"$ifNull": [{"$arrayElemAt":
    ["$totals.orphans", 0]}, 0]}, "remaining_deletable": {"$ifNull":
    [{"$arrayElemAt": ["$totals.deletable", 0]}, 0]}, "worst": "$worst"}}]') |
    first | toJson }}

  delete_batches: >-
    {{ {'ids': outputs.detect.rows[0].orphan_ids, 'n': inputs.batch_size} |
    jq('.n as $n | [.ids | to_entries | group_by(.key / $n | floor) | .[] |
    map({"$oid": .value}) | tojson]') | first | toJson }}

tasks:
  - id: check_inputs
    type: io.kestra.plugin.core.flow.If
    description: >-
      Refuse a child and parent collection that are the same, a batch_size or
      delete_limit below 1, a negative age gate or an orphan share ceiling
      outside 0 to 1. Each of those makes the delete undefined, so the run stops
      before a single read.
    condition: "{{ inputs.child_collection == inputs.parent_collection or
      inputs.batch_size < 1 or inputs.delete_limit < 1 or
      inputs.min_orphan_age_minutes < 0 or inputs.max_orphan_ratio <= 0 or
      inputs.max_orphan_ratio > 1 }}"
    then:
      - id: bad_inputs
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          Input check failed. child_collection={{ inputs.child_collection }}
          parent_collection={{ inputs.parent_collection }} batch_size={{
          inputs.batch_size }} delete_limit={{ inputs.delete_limit }}
          min_orphan_age_minutes={{ inputs.min_orphan_age_minutes }}
          max_orphan_ratio={{ inputs.max_orphan_ratio }}. The two collections
          must differ, batch_size and delete_limit must be at least 1, the age
          gate cannot be negative and the orphan share ceiling must sit in 0 to
          1. Nothing was read. Nothing was written.

  - id: type_probe
    type: io.kestra.plugin.mongodb.Aggregate
    description: >-
      Read-only. Groups the reference field by its BSON type on the child side,
      then unions the same grouping of the parent key field. A `$lookup`
      compares types strictly, so a string reference against an ObjectId key
      matches nothing at all. Without this probe such a join reports every
      referencing child as an orphan.
    connection:
      uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
        secret('MONGODB_URI') }}"
    database: "{{ inputs.database }}"
    collection: "{{ inputs.child_collection }}"
    pipeline: "{{ render(vars.type_probe_pipeline) }}"
    store: FETCH

  - id: log_types
    type: io.kestra.plugin.core.log.Log
    description: Print the join compatibility verdict, because task outputs are not
      in the execution JSON.
    message: >-
      Join types for {{ inputs.child_collection }}.{{ inputs.reference_field }}
      against {{ inputs.parent_collection }}.{{ inputs.parent_key_field }}: {{
      outputs.type_probe.rows ?? 'none' }}. Types disjoint: {{
      render(vars.type_mismatch) }}. A disjoint pair means the join can never
      match, so no delete is allowed.

  - id: detect
    type: io.kestra.plugin.mongodb.Aggregate
    description: >-
      Read-only. Joins every child that carries the reference to the parent
      collection with `$lookup`, marks a child whose join array is empty as
      dangling, derives when each child was written, then splits the dangling
      set by the age gate. One `$facet` returns the exact referencing total, the
      exact orphan count, how many are past the age gate, how many were spared
      as fresh, the capped id list the delete acts on and a sample of the oldest
      orphans.
    connection:
      uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
        secret('MONGODB_URI') }}"
    database: "{{ inputs.database }}"
    collection: "{{ inputs.child_collection }}"
    pipeline: "{{ render(vars.detect_pipeline) }}"
    store: FETCH

  - id: report
    type: io.kestra.plugin.core.log.Log
    description: Log the counts the gates below read. The age cutoff is printed so a
      spared child can be checked by hand.
    message: >-
      Checked {{ inputs.database }}.{{ inputs.child_collection }} references to
      {{ inputs.parent_collection }}.{{ inputs.parent_key_field }}. Children
      carrying a reference: {{ outputs.detect.rows[0].referencing ?? 0 }}.
      Dangling: {{ outputs.detect.rows[0].orphans ?? 0 }}. Past the {{
      inputs.min_orphan_age_minutes }} minute age gate: {{
      outputs.detect.rows[0].deletable ?? 0 }}. Spared as too fresh to judge: {{
      outputs.detect.rows[0].spared_fresh ?? 0 }}. Targeted this run: {{
      outputs.detect.rows[0].targeted ?? 0 }} (delete_limit {{
      inputs.delete_limit }}). Orphan share {{
      outputs.detect.rows[0].orphan_ratio ?? 0 }} against ceiling {{
      inputs.max_orphan_ratio }}. Age cutoff {{ render(vars.cutoff_iso) }}.
      Oldest orphans: {{ outputs.detect.rows[0].sample ?? 'none' }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    description: A dangling reference means alert and consider the delete. A
      collection whose references all resolve just logs and stops.
    condition: "{{ (outputs.detect.rows[0].orphans ?? 0) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': ('MongoDB orphan guard on ' ~ inputs.database ~ '.' ~ inputs.child_collection ~ ': ' ~ (outputs.detect.rows[0].orphans ?? 0) ~ ' document(s) reference a missing ' ~ inputs.parent_collection ~ ' parent through ' ~ inputs.reference_field ~ '. ' ~ (outputs.detect.rows[0].deletable ?? 0) ~ ' past the age gate, ' ~ (outputs.detect.rows[0].spared_fresh ?? 0) ~ ' spared as fresh. Orphan share ' ~ (outputs.detect.rows[0].orphan_ratio ?? 0) ~ '. Execution ' ~ execution.id ~ '.')} | toJson }}

      - id: planned
        type: io.kestra.plugin.core.log.Log
        description: What a remediation run would do, printed on every run including a
          dry one.
        message: >-
          A cleanup would delete {{ outputs.detect.rows[0].targeted ?? 0 }} aged
          orphan(s) from {{ inputs.database }}.{{ inputs.child_collection }} in
          batches of {{ inputs.batch_size }}, by _id only. {{
          outputs.detect.rows[0].spared_fresh ?? 0 }} dangling child(ren) stay
          because they are younger than {{ inputs.min_orphan_age_minutes }}
          minutes. dry_run={{ inputs.dry_run }} auto_remediate={{
          inputs.auto_remediate }}. A delete runs only when auto_remediate is
          true and dry_run is false and a reviewer approves the pause.

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        description: The delete opens only when auto_remediate is true and dry_run is false.
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: safety_check
            type: io.kestra.plugin.core.flow.If
            description: >-
              Stop the write when the join looks broken rather than the data.
              Disjoint BSON types on the two sides of the join read as a join
              that matches nothing. So does an orphan share above the ceiling.
              Deleting on that evidence would empty the collection.
            condition: "{{ render(vars.type_mismatch) == 'true' or
              (outputs.detect.rows[0].orphan_ratio ?? 0) >
              inputs.max_orphan_ratio }}"
            then:
              - id: refuse_delete
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: >-
                  Refusing to delete from {{ inputs.database }}.{{
                  inputs.child_collection }}. Join types disjoint: {{
                  render(vars.type_mismatch) }}. Orphan share {{
                  outputs.detect.rows[0].orphan_ratio ?? 0 }} against ceiling {{
                  inputs.max_orphan_ratio }}. Types seen: {{
                  outputs.type_probe.rows ?? 'none' }}. The reference field and
                  the parent key may hold different BSON types, which leaves the
                  join incapable of matching. Or so much of the collection looks
                  orphaned that the join is the likelier fault. Fix the
                  reference types or raise max_orphan_ratio deliberately.
                  Nothing was deleted.
            else:
              - id: hold
                type: io.kestra.plugin.core.flow.Pause
                description: >-
                  The delete is irreversible, so a human answers first. With no
                  answer nothing is deleted and the findings stay open.
                pauseDuration: "{{ inputs.approval_timeout }}"
                onResume:
                  - id: approved
                    type: BOOL
                    displayName: Delete the aged orphan documents?
                    defaults: false
                  - id: note
                    type: STRING
                    displayName: Decision note
                    defaults: ""

              - id: apply
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.hold.onResume.approved ?? false }}"
                then:
                  - id: delete_branch
                    type: io.kestra.plugin.core.flow.If
                    description: Only open the write path when the detection targeted at least one
                      aged orphan.
                    condition: "{{ (outputs.detect.rows[0].targeted ?? 0) > 0 }}"
                    then:
                      - id: remediate
                        type: io.kestra.plugin.core.flow.Loop
                        description: >-
                          One batch of orphan ids per iteration.
                          concurrencyLimit 1 keeps the batches sequential,
                          because parallel deletes on one collection race for no
                          gain.
                        concurrencyLimit: 1
                        values: "{{ render(vars.delete_batches) }}"
                        tasks:
                          - id: purge
                            type: io.kestra.plugin.mongodb.Delete
                            description: >-
                              One DELETE_MANY per batch, filtered on the
                              targeted _id list only. It never filters on the
                              reference field, so a child whose parent exists
                              cannot be caught by a stale filter, nor can a
                              fresh child the age gate spared.
                            connection:
                              uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
                                secret('MONGODB_URI') }}"
                            database: "{{ inputs.database }}"
                            collection: "{{ inputs.child_collection }}"
                            operation: DELETE_MANY
                            filter: "{{ {'ids': item.value} | jq('{\"_id\": {\"$in\": (.ids | fromjson)}}')
                              | first | toJson }}"

                      - id: log_deleted
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          Delete loop finished for {{
                          outputs.detect.rows[0].targeted ?? 0 }} aged
                          orphan(s). The re-join below is the proof, not this
                          count.

                      - id: reverify
                        type: io.kestra.plugin.mongodb.Aggregate
                        description: >-
                          Read-only. Runs the same `$lookup` over the whole
                          child collection again, against the same frozen age
                          cutoff, so a child a writer inserted while the run was
                          paused is counted rather than hidden.
                        connection:
                          uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
                            secret('MONGODB_URI') }}"
                        database: "{{ inputs.database }}"
                        collection: "{{ inputs.child_collection }}"
                        pipeline: "{{ render(vars.reverify_pipeline) }}"
                        store: FETCH

                      - id: log_reverify
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          Re-join after the delete found {{
                          outputs.reverify.rows[0].remaining_orphans ?? 0 }}
                          dangling reference(s) left, of which {{
                          outputs.reverify.rows[0].remaining_deletable ?? 0 }}
                          are past the age gate. Children still carrying a
                          reference: {{
                          outputs.reverify.rows[0].still_referencing ?? 0 }}.
                          Worst remaining: {{ outputs.reverify.rows[0].worst ??
                          'none' }}.

                      - id: verify_gate
                        type: io.kestra.plugin.core.flow.If
                        description: Fail the run if any orphan past the age gate is still there after
                          the delete.
                        condition: "{{ (outputs.reverify.rows[0].remaining_deletable ?? 0) > 0 }}"
                        then:
                          - id: remediation_failed
                            type: io.kestra.plugin.core.execution.Fail
                            errorMessage: >-
                              The delete ran but the re-join still finds {{
                              outputs.reverify.rows[0].remaining_deletable ?? 0
                              }} aged orphan(s) in {{ inputs.database }}.{{
                              inputs.child_collection }}. Causes to check: more
                              were aged than delete_limit {{ inputs.delete_limit
                              }} allowed; a writer added another child for a
                              missing parent during the run; the delete was
                              refused by the server. A refused write still
                              reports success, so this re-join is the only thing
                              that decides.
                        else:
                          - id: remediated
                            type: io.kestra.plugin.core.log.Log
                            message: >-
                              Cleared {{ outputs.detect.rows[0].targeted ?? 0 }}
                              aged orphan(s) from {{ inputs.database }}.{{
                              inputs.child_collection }}. The re-join finds {{
                              outputs.reverify.rows[0].remaining_deletable ?? 0
                              }} aged orphan(s) left and {{
                              outputs.reverify.rows[0].remaining_orphans ?? 0 }}
                              dangling reference(s) in total, the difference
                              being children too fresh to judge. Every remaining
                              reference that is old enough to judge now resolves
                              to a parent.
                    else:
                      - id: nothing_to_delete
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          {{ outputs.detect.rows[0].orphans ?? 0 }} dangling
                          reference(s) were reported but none is past the age
                          gate, so nothing was deleted. Every one of them was
                          written in the last {{ inputs.min_orphan_age_minutes
                          }} minutes. Lower min_orphan_age_minutes only if your
                          writers always insert the parent first.
                else:
                  - id: declined
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Approval not granted{{ (outputs.hold.onResume.note ?? '')
                      == '' ? '' : (', note: ' ~ outputs.hold.onResume.note) }}.
                      No document was deleted. The dangling references stay in
                      place.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: >-
          No dangling references. All {{ outputs.detect.rows[0].referencing ?? 0
          }} child document(s) in {{ inputs.database }}.{{
          inputs.child_collection }} that carry {{ inputs.reference_field }}
          resolve to a {{ inputs.parent_collection }} parent.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': ('MongoDB orphan guard failed on ' ~ inputs.database ~ '.' ~ inputs.child_collection ~ ' (execution ' ~ execution.id ~ '). Check the logs.')} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: >-
      Shipped disabled. Its inputs force safe mode, so even an enabled schedule
      only ever reports and never deletes on its own.
    cron: "0 3 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

MongoDB has no foreign keys. A manual reference is just a field holding another document's key, so deleting a parent is allowed while children still point at it. The moment a customer record goes, every order carrying that customerId references a key nothing resolves. Nothing errors. The orders still read back fine on their own. The damage shows up in the joins: a $lookup on those orders returns an empty array, an aggregation that unwinds the join drops the row entirely, a revenue report grouped by customer quietly loses the money. Deletes land through an admin console, a GDPR erasure job, a retention sweep or a support script, so the children outlive their parent in almost every application that models references by hand.

This blueprint is the repair for that, inside MongoDB itself. It joins the child collection to the parent collection server side with $lookup, treats an empty join array as a dangling reference, then splits the dangling set by age. A child written in the last few minutes is reported but never deleted, because a child can legitimately be written just before its parent or land in a multi document insert that is still in flight. The aged orphans are deleted by _id behind a safe mode gate and an approval. Then the same join runs again. If an orphan old enough to judge is still there, the run fails.

This blueprint was created by zkasuran.

Safety

Read-only by default. The first run with the defaults only counts and reports. A delete needs auto_remediate true, dry_run false, then a human answering the Pause. Three conditions, all of them, before a document is removed.
The age gate is the part that stops a race from becoming data loss. A dangling child younger than min_orphan_age_minutes is counted, listed, never deleted. Without it a sweep running during a multi document insert deletes rows whose parent was about to arrive.
A $lookup compares BSON types strictly. A string customerId against an ObjectId _id matches nothing, which makes every referencing child look orphaned. type_probe groups both sides by BSON type first and the write path refuses outright when the two type sets are disjoint.
max_orphan_ratio is the second rail on the same risk. When more than half the referencing children look orphaned, the likelier fault is the join, not the data, so the run fails instead of deleting.
The delete filters on the targeted _id list, never on the reference field. A child whose parent exists cannot be caught by a stale filter. Neither can a fresh child the age gate spared.
A child missing the reference field or holding null in it, is never read as an orphan. Only documents that actually claim a parent are judged.
The delete is capped at delete_limit per run and batched by batch_size. If more are aged than the cap, the re-join fails until a later run finishes, so a partial clear is never reported as done.
What a wrong delete costs: the child documents are gone. There is no undo on a DELETE_MANY, so on orders that is revenue history with no backup behind it. That is why the type probe, the share ceiling, the age gate, auto_remediate, dry_run and the Pause all sit in front of it.

How it works

check_inputs (io.kestra.plugin.core.flow.If) fails the run before any read when the two collections are the same, a limit is below 1, the age gate is negative or the orphan share ceiling sits outside 0 to 1.
type_probe (io.kestra.plugin.mongodb.Aggregate) groups the reference field by BSON type on the child side, then $unionWith adds the same grouping of the parent key field. This is the join compatibility check.
log_types (io.kestra.plugin.core.log.Log) prints both type sets with the disjoint verdict, because task outputs are not in the execution JSON.
detect (io.kestra.plugin.mongodb.Aggregate) runs the $lookup from the child collection into the parent collection, marks a child with an empty join array as dangling, derives the child write time from age_field or from the timestamp inside its ObjectId, then compares it with a cutoff frozen on execution.startDate. One $facet returns the referencing total, the orphan count, the aged count, the spared count, the capped id list the delete acts on, the orphan share and a sample of the oldest orphans.
report (io.kestra.plugin.core.log.Log) logs those numbers with the cutoff, so a spared child can be checked by hand.
route (io.kestra.plugin.core.flow.If): zero orphans logs clean and stops. One or more sends alert to Slack then prints planned, which states what a remediation run would do on every run including a dry one.
remediate_gate (io.kestra.plugin.core.flow.If) opens only with auto_remediate true and dry_run false. Inside it safety_check fails the run through io.kestra.plugin.core.execution.Fail on disjoint join types or an orphan share above the ceiling. Otherwise hold (io.kestra.plugin.core.flow.Pause) waits for an answer.
On approval remediate (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) walks the id batches and purge (io.kestra.plugin.mongodb.Delete, DELETE_MANY) removes each batch by _id. With nothing aged the run logs nothing_to_delete.
reverify (io.kestra.plugin.mongodb.Aggregate) runs the join again against the same frozen cutoff and verify_gate fails the run if any aged orphan survived. Without approval declined records the decision and nothing changes.
errors posts to Slack when any task fails, for example when MongoDB is unreachable.

What you get

The exact number of child documents whose reference resolves to nothing, which no MongoDB command reports for you, because the database never knew the reference existed.
A Slack alert the moment a reference goes dangling, before a report joining on it starts dropping rows.
An age gate that separates a real orphan from a write still in flight, so a cleanup running next to a live writer does not delete a child whose parent is one insert behind.
A join compatibility probe that catches the mismatch that makes a naive orphan sweep delete the whole collection.
A gated, approved delete of the aged orphans by _id, capped per run.
A re-join that fails the run when an aged orphan survived, so a refused write or a concurrent writer is caught rather than hidden.

Who it's for

Teams modelling one to many with a manual reference, which is most MongoDB applications, with deletes arriving from a console, an erasure request or a retention job.
Data engineers whose $lookup aggregations lose rows, where the join looks correct because the missing parent is invisible from the child side.
Anyone preparing a collection for stricter integrity checks who has to find out how much already dangles.

Why orchestrate this with Kestra

Referential integrity in MongoDB is the application's job, which means nobody owns it once the application that wrote the reference has moved on. The usual answer is a script somebody runs after an incident. This flow makes it scheduled, reviewable work. The detection is one server side pipeline, so a large collection never leaves the database. The age gate turns a race into a documented decision. The delete sits behind auto_remediate, dry_run, a type probe, a share ceiling then a Pause with the answer recorded on the execution. The batches are serialised by concurrencyLimit: 1. The re-join turns the delete into a checked outcome rather than a hopeful one. The Slack alert and the error path are the same ones the rest of your flows use.

Tested end to end

On Kestra 2.0.5 OSS against a MongoDB 7.0 container, on the final file. appdata.customers held 3 parents, appdata.orders held 8 children whose customerId all resolved. Deleting one customer orphaned 3 of those orders. A fourth orphan was then written fresh, with no parent at all, to test the age gate. Every before and after number was read back with mongosh, not only from the Kestra log.

Run	Inputs	Result
Clean	defaults on intact data	8 referencing, 0 dangling. `clean` logged, no alert, SUCCESS
Detect	defaults after deleting one customer	3 dangling, 3 aged, 0 spared, share 0.375. Slack alert, SUCCESS
Dry run	the same	Reports the delete it would make. `mongosh` still 8 orders, 3 orphans
Armed but dry	`auto_remediate: true`, `dry_run: true`	Gate stayed closed, no `Pause`. `mongosh` unchanged
Declined	`auto_remediate: true`, `dry_run: false`, resumed `approved: false`	Paused, then `declined`. `mongosh` still 8 orders, 3 orphans
Age gate	a fourth orphan written fresh, `min_orphan_age_minutes: 15`, approved	3 aged orphans deleted, the fresh one spared. `mongosh` 9 orders with 4 orphans before, 6 orders with 1 orphan after
Nothing aged	only the fresh orphan left, approved	`nothing_to_delete`, no delete issued at all
Share ceiling	a second parent deleted, `max_orphan_ratio: 0.5`	Share 0.6667, `refuse_delete` failed the run before the `Pause`, nothing deleted
Re-verify fails	`max_orphan_ratio: 0.9`, `delete_limit: 1` with 3 aged orphans, approved	1 deleted, re-join found 2 aged, `verify_gate` failed the run, `errors` alerted. `mongosh` 6 orders to 5
Type mismatch	a child collection whose reference is a string against an `ObjectId` key	Types disjoint, `refuse_delete` failed the run, all 3 documents still present
Write time field	`age_field: createdAt` instead of the `ObjectId` timestamp	Same 2 aged plus 1 spared split, read from `createdAt` to the millisecond
Full clear	`min_orphan_age_minutes: 0`, approved	All 3 orphans deleted, re-join 0, SUCCESS. `mongosh` 5 orders to 2
Bad input	`max_orphan_ratio: 1.5`	`check_inputs` failed the run before any read
Unreachable	`connection_uri` pointed at a dead host	`type_probe` failed, `errors` posted to Slack, execution FAILED

Does the detector actually fail on orphans

The clean run and the detect run isolate it. Nothing changed between them except one deleted parent: the intact collection returned 0 dangling out of 8 referencing children, the same flow with the same inputs then returned 3. mongosh confirmed both by running the same $lookup plus $match on an empty join itself. The total document count is the other half of the proof: 9 orders with 4 orphans before the age gate run, 6 orders with 1 orphan after, so exactly the 3 aged orphans went while the fresh one stayed. The re-verify gate was proven the other way by capping delete_limit at 1, which left 2 aged orphans behind then failed the run.

Prerequisites

A reachable MongoDB instance and a user that can read both collections then run a delete on the child collection.
A child collection whose _id is a stock ObjectId. The delete round trips ids through $oid. With no age_field the write time comes from the timestamp inside the ObjectId.
An index on the parent key field, normally _id, so the join is served without a collection scan. An index on the child reference field helps the sample sorts.
Local testing: docker run -d --name mdb --network mynet mongo:7.0, seed a parent collection plus a child collection referencing it, delete one parent, then run the flow with connection_uri: mongodb://mdb:27017.

Secrets

MONGODB_URI: the MongoDB connection string. Read when connection_uri is left empty, which is how the flow should run in production. In Kestra OSS, provide it as the base64-encoded environment variable SECRET_MONGODB_URI.
SLACK_WEBHOOK_URL: used by alert and the errors block. Provide it as SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Default	Purpose
`connection_uri`	empty	Plain connection string for a local test. Empty reads `MONGODB_URI`
`database`	`appdata`	The database holding both collections
`child_collection`	`orders`	The collection that carries the references
`parent_collection`	`customers`	The collection the references point at
`reference_field`	`customerId`	The field on a child holding the parent key
`parent_key_field`	`_id`	The field on a parent the reference resolves against
`age_field`	empty	Child write time field. Empty takes it from the `ObjectId` `_id`
`min_orphan_age_minutes`	`15`	A dangling child younger than this is reported, never deleted
`max_orphan_ratio`	`0.5`	Refuse to delete above this orphan share of referencing children
`scan_limit`	`25`	How many orphans the report lists. Counts stay exact
`delete_limit`	`50000`	Ceiling on aged orphans deleted per run
`batch_size`	`500`	Orphan ids per delete batch
`approval_timeout`	`PT2H`	How long the pause waits for an answer
`auto_remediate`	`false`	Allow the delete
`dry_run`	`true`	Must be `false` for a delete to apply

Quick start

Add the MONGODB_URI and SLACK_WEBHOOK_URL secrets.
Set database, the two collection names, reference_field then parent_key_field. Run the flow. It reports the dangling references, how many are old enough to judge, the orphan share. It deletes nothing.
Read log_types first. If the two BSON type sets are disjoint, fix the reference types. There is nothing to clean up, the join itself is wrong.
Set auto_remediate: true with dry_run: false, run it again, then answer the pause. The aged orphans are deleted and the re-join proves none is left.
Enable the daily schedule, which always runs in safe mode.

How to extend

Archive before deleting: add a $merge of the targeted orphans into a quarantine collection ahead of the delete, so a wrong parent deletion is recoverable.
Repair instead of delete: swap purge for io.kestra.plugin.mongodb.Update with UPDATE_MANY to point the orphans at a placeholder parent, where the child is worth keeping.
Guard several reference pairs by wrapping the tasks in a io.kestra.plugin.core.flow.Loop over a list of child, parent, field triples, with concurrencyLimit: 1 on the part that deletes.
Run it from the deletion side: trigger the flow after the job that deletes parents, so the children are cleaned in the same window.
Replace the Pause with a Slack approval. Or drop it for a collection you are sure about.

MongoDB Dangling Reference Orphan Cleanup Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Does the detector actually fail on orphans

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

MongoDB Dangling Reference Orphan Cleanup Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Does the detector actually fail on orphans

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

More Infrastructure Blueprints

New to Kestra?