MongoDB Duplicate Document Dedup Guard with Kestra

id: mongodb-duplicate-document-dedup-guard
namespace: company.team

description: |
  Find the MongoDB documents that share a business key no unique index could keep apart. The
  bundled MongoDB plugin cannot create an index, so a collection cannot be given a unique
  constraint from Kestra. Nothing then stops an at-least-once ingest or a retried upsert from
  writing the same logical record twice. Duplicates build up on an orderId, an email or an
  external id and every reader that assumes the key is unique double counts or picks an
  arbitrary copy. This flow groups the collection by the business key, reports the keys carried
  by more than one document, keeps one survivor per key by a rule you choose, deletes the surplus
  copies behind a safe mode gate and an approval, then groups the collection again and fails the
  run if any key still has more than one document.

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: collection
    type: STRING
    displayName: Collection
    defaults: orders

  - id: business_key
    type: STRING
    displayName: Business key
    description: >-
      The field that should identify one logical record, for example email or
      orderId. For a composite key pass the fields comma separated, for example
      orderId,source. A document that is missing any key field or holds null in
      it, is never grouped and never touched.
    defaults: email

  - id: survivor_strategy
    type: SELECT
    displayName: Which copy to keep
    description: >-
      How the one survivor per key is chosen. oldest_id keeps the smallest _id,
      which for a stock ObjectId is the first inserted. newest_id keeps the
      largest _id. newest_timestamp keeps the document with the largest
      timestamp_field, with the smallest _id as the tie break. The surplus is
      every other copy of that key.
    values:
      - oldest_id
      - newest_id
      - newest_timestamp
    defaults: oldest_id

  - id: timestamp_field
    type: STRING
    displayName: Timestamp field
    description: >-
      The BSON date field used only when survivor_strategy is newest_timestamp.
      Ignored by the other strategies.
    defaults: createdAt

  - id: scan_limit
    type: INT
    displayName: Duplicate keys named per report
    description: >-
      Ceiling on how many duplicate keys the report and the re-check name. The
      counts are exact and server-side, so the cap only bounds how many keys are
      listed.
    defaults: 25

  - id: delete_limit
    type: INT
    displayName: Surplus documents deleted per run
    description: >-
      Ceiling on how many surplus copies one run deletes, so a huge backlog
      cannot build one giant write. The detection count is exact. If the surplus
      is larger than this the run clears part of it and the re-check 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, so a
      large dedup never becomes one long running write.
    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:
  detect_pipeline: >-
    {{ {'bk': inputs.business_key, 'sl': inputs.scan_limit, 'dl':
    inputs.delete_limit, 'strat': inputs.survivor_strategy, 'ts':
    inputs.timestamp_field} | jq('.bk as $bk | .ts as $ts | .strat as $strat |
    .sl as $sl | .dl as $dl | ($bk | split(",") | map(gsub(" ";"")) |
    map(select(. != ""))) as $fields | {"$and": ($fields | map({(.): {"$exists":
    true, "$ne": null}}))} as $exist | ($fields | to_entries | reduce .[] as $e
    ({}; . + {("k" + ($e.key | tostring)): ("$" + $e.value)})) as $gid | (if
    $strat == "newest_id" then {"_id": -1} elif $strat == "newest_timestamp"
    then {($ts): -1, "_id": 1} else {"_id": 1} end) as $sort | [{"$match":
    $exist}, {"$sort": $sort}, {"$group": {"_id": $gid, "count": {"$sum": 1},
    "survivor_id": {"$first": "$_id"}, "all_ids": {"$push": "$_id"}}},
    {"$match": {"count": {"$gt": 1}}}, {"$set": {"surplus_ids": {"$filter":
    {"input": "$all_ids", "as": "d", "cond": {"$ne": ["$$d",
    "$survivor_id"]}}}}}, {"$facet": {"sample": [{"$project": {"_id": 0, "key":
    "$_id", "copies": "$count", "survivor": {"$toString": "$survivor_id"},
    "surplus": {"$size": "$surplus_ids"}}}, {"$sort": {"copies": -1}},
    {"$limit": $sl}], "surplus": [{"$unwind": "$surplus_ids"}, {"$limit": $dl},
    {"$group": {"_id": "all", "ids": {"$push": {"$toString":
    "$surplus_ids"}}}}], "totals": [{"$group": {"_id": "all", "dup_keys":
    {"$sum": 1}, "surplus_total": {"$sum": {"$size": "$surplus_ids"}}}}]}},
    {"$project": {"dup_keys": {"$ifNull": [{"$arrayElemAt": ["$totals.dup_keys",
    0]}, 0]}, "surplus_total": {"$ifNull": [{"$arrayElemAt":
    ["$totals.surplus_total", 0]}, 0]}, "surplus_ids": {"$ifNull":
    [{"$arrayElemAt": ["$surplus.ids", 0]}, []]}, "sample": "$sample"}},
    {"$set": {"surplus_targeted": {"$size": "$surplus_ids"}}}]') | first |
    toJson }}

  reverify_pipeline: >-
    {{ {'bk': inputs.business_key} | jq('.bk as $bk | ($bk | split(",") |
    map(gsub(" ";"")) | map(select(. != ""))) as $fields | {"$and": ($fields |
    map({(.): {"$exists": true, "$ne": null}}))} as $exist | ($fields |
    to_entries | reduce .[] as $e ({}; . + {("k" + ($e.key | tostring)): ("$" +
    $e.value)})) as $gid | [{"$match": $exist}, {"$group": {"_id": $gid,
    "count": {"$sum": 1}}}, {"$match": {"count": {"$gt": 1}}}, {"$facet":
    {"dups": [{"$count": "n"}], "worst": [{"$sort": {"count": -1}}, {"$limit":
    5}, {"$project": {"_id": 0, "key": "$_id", "copies": "$count"}}]}},
    {"$project": {"remaining_dup_keys": {"$ifNull": [{"$arrayElemAt":
    ["$dups.n", 0]}, 0]}, "worst": "$worst"}}]') | first | toJson }}

  delete_batches: >-
    {{ {'ids': outputs.detect.rows[0].surplus_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 newest_timestamp strategy with no timestamp_field and refuse a
      batch_size or delete_limit below 1, before a single read. Each would make
      the delete loop undefined.
    condition: "{{ (inputs.survivor_strategy == 'newest_timestamp' and
      inputs.timestamp_field == '') or inputs.batch_size < 1 or
      inputs.delete_limit < 1 }}"
    then:
      - id: bad_inputs
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          Input check failed. survivor_strategy={{ inputs.survivor_strategy }}
          timestamp_field={{ inputs.timestamp_field }} batch_size={{
          inputs.batch_size }} delete_limit={{ inputs.delete_limit }}.
          newest_timestamp needs a timestamp_field. batch_size and delete_limit
          must each be at least 1. Nothing was read. Nothing was written.

  - id: detect
    type: io.kestra.plugin.mongodb.Aggregate
    description: >-
      Read-only. Groups the collection by the business key, keeps the chosen
      survivor per key with a sort then a group first and marks every other copy
      as surplus by its _id. One facet then returns the exact number of
      duplicate keys, the exact surplus total, the capped list of surplus ids
      the delete acts on and the worst scan_limit keys by copy count. The
      survivor id is never in the surplus list, so a delete can never remove the
      keeper.
    connection:
      uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
        secret('MONGODB_URI') }}"
    database: "{{ inputs.database }}"
    collection: "{{ inputs.collection }}"
    pipeline: "{{ render(vars.detect_pipeline) }}"
    store: FETCH

  - id: report
    type: io.kestra.plugin.core.log.Log
    description: Log the counts. Task outputs are not in the execution JSON, so the
      numbers the gates use are written here.
    message: >-
      Checked {{ inputs.database }}.{{ inputs.collection }} for documents
      sharing business key {{ inputs.business_key }}. Duplicate keys: {{
      outputs.detect.rows[0].dup_keys ?? 0 }}. Surplus copies in total: {{
      outputs.detect.rows[0].surplus_total ?? 0 }}. Targeted this run: {{
      outputs.detect.rows[0].surplus_targeted ?? 0 }} (delete_limit {{
      inputs.delete_limit }}). Survivor kept per key by {{
      inputs.survivor_strategy }}. Worst keys: {{ outputs.detect.rows[0].sample
      ?? 'none' }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    description: A duplicate key means alert and consider the delete. A collection
      with one document per key just logs and stops.
    condition: "{{ (outputs.detect.rows[0].dup_keys ?? 0) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': ('MongoDB dedup guard on ' ~ inputs.database ~ '.' ~ inputs.collection ~ ': ' ~ (outputs.detect.rows[0].dup_keys ?? 0) ~ ' business key(s) carried by more than one document on ' ~ inputs.business_key ~ ', ' ~ (outputs.detect.rows[0].surplus_total ?? 0) ~ ' surplus copy(s). Survivor kept by ' ~ inputs.survivor_strategy ~ '. 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 dedup would keep one document per key by {{ inputs.survivor_strategy
          }} and delete {{ outputs.detect.rows[0].surplus_targeted ?? 0 }}
          surplus copy(s) in batches of {{ inputs.batch_size }}. 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: hold
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              The delete is destructive, 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 surplus duplicate 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 found a surplus to
                  remove.
                condition: "{{ (outputs.detect.rows[0].surplus_targeted ?? 0) > 0 }}"
                then:
                  - id: remediate
                    type: io.kestra.plugin.core.flow.Loop
                    description: >-
                      One batch of surplus ids per iteration. concurrencyLimit 1
                      keeps the batches sequential, because parallel deletes on
                      the same 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 surplus _id
                          list only. It never filters on the business key, which
                          would match the survivor too, so the keeper is safe by
                          construction.
                        connection:
                          uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
                            secret('MONGODB_URI') }}"
                        database: "{{ inputs.database }}"
                        collection: "{{ inputs.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].surplus_targeted ?? 0 }} surplus
                      copy(s) across {{ outputs.detect.rows[0].dup_keys ?? 0 }}
                      key(s). The re-group below is the proof, not this count.

                  - id: reverify
                    type: io.kestra.plugin.mongodb.Aggregate
                    description: >-
                      Read-only. Groups the whole collection by the business key
                      again and counts how many keys still carry more than one
                      document, collection wide, so a copy a writer inserted
                      while the run was paused is caught rather than hidden.
                    connection:
                      uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
                        secret('MONGODB_URI') }}"
                    database: "{{ inputs.database }}"
                    collection: "{{ inputs.collection }}"
                    pipeline: "{{ render(vars.reverify_pipeline) }}"
                    store: FETCH

                  - id: log_reverify
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Re-check after the delete found {{
                      outputs.reverify.rows[0].remaining_dup_keys ?? 0 }}
                      business key(s) still carried by more than one document.
                      Worst remaining: {{ outputs.reverify.rows[0].worst ??
                      'none' }}.

                  - id: verify_gate
                    type: io.kestra.plugin.core.flow.If
                    description: Fail the run if any business key still has more than one document
                      after the delete.
                    condition: "{{ (outputs.reverify.rows[0].remaining_dup_keys ?? 0) > 0 }}"
                    then:
                      - id: remediation_failed
                        type: io.kestra.plugin.core.execution.Fail
                        errorMessage: >-
                          The delete ran but the re-group still finds {{
                          outputs.reverify.rows[0].remaining_dup_keys ?? 0 }}
                          business key(s) in {{ inputs.database }}.{{
                          inputs.collection }} carried by more than one
                          document. The surplus was larger than delete_limit {{
                          inputs.delete_limit }} or a writer inserted another
                          copy during the run. The key is still not unique.
                    else:
                      - id: remediated
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          Removed the surplus for {{
                          outputs.detect.rows[0].dup_keys ?? 0 }} key(s) in {{
                          inputs.database }}.{{ inputs.collection }}. The
                          re-group finds {{
                          outputs.reverify.rows[0].remaining_dup_keys ?? 0 }}
                          duplicate key(s) left, so every grouped key now
                          resolves to a single document.
                else:
                  - id: nothing_to_delete
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      {{ outputs.detect.rows[0].dup_keys ?? 0 }} duplicate
                      key(s) were reported but the targeted surplus is 0, so
                      nothing was deleted. Raise delete_limit if a cap held the
                      surplus back.
            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 duplicate copies stay in place.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: >-
          No duplicates. Every business key {{ inputs.business_key }} in {{
          inputs.database }}.{{ inputs.collection }} is carried by a single
          document.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': ('MongoDB dedup guard failed on ' ~ inputs.database ~ '.' ~ inputs.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 2 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

MongoDB keeps a key unique with a unique index. The bundled Kestra MongoDB plugin has no task that creates an index, so a collection cannot be given that constraint from a Kestra flow at all. Nothing then stops the same logical record from being written twice: an at-least-once ingest replays a message, a retried upsert runs again after a timeout, a backfill reloads a file that was already loaded. The second write lands as a new document because the key it should have collided on was never made unique. Over weeks the duplicates pile up on an orderId, an email or an external id and every reader that assumes one document per key quietly double counts a total or picks an arbitrary copy.

This blueprint is the outside repair for that. It groups the collection by a business key you pass in, so it works on any collection and reports the keys carried by more than one document. It keeps one survivor per key by a rule you choose, the oldest _id, the newest _id or the newest by a timestamp field and marks every other copy as surplus. Behind a safe mode gate and an approval it deletes the surplus by _id, never by the key match, so the survivor is safe by construction. Then it groups the collection again and fails the run if any key still has more than one document.

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 delete filters on the surplus _id list, never on the business key. A key match would select the survivor too. The survivor _id is excluded from the surplus list when it is built, so the keeper cannot be deleted even if the delete and the detection disagree.
The survivor is deterministic, not arbitrary. A sort before the group fixes which copy is first, so the same inputs keep the same document every run. Picking at random is how a dedup deletes the record somebody needed.
A document missing any business key field or holding null in it, is never grouped and never touched. Dedup applies only to documents that actually carry the key.
The delete is capped at delete_limit per run and batched by batch_size, so a large backlog cannot become one giant write. If the surplus is larger than the cap, the re-group fails until a later run finishes, so a partial clear is never reported as done.
The re-group is the proof, not the delete count. If any key still has more than one document, the run fails rather than reporting success.

How it works

check_inputs (io.kestra.plugin.core.flow.If) fails the run before any read on a newest_timestamp strategy with no timestamp_field or a batch_size or delete_limit below 1.
detect (io.kestra.plugin.mongodb.Aggregate) matches documents that carry the whole business key, sorts them so the chosen survivor is first, groups by the key with $first capturing the survivor _id and $push capturing all _ids, then keeps only the groups with a count above 1. A $filter marks every id except the survivor as surplus. One $facet returns the exact duplicate key count, the exact surplus total, the capped surplus id list and the worst scan_limit keys.
report (io.kestra.plugin.core.log.Log) logs those numbers, because task outputs are not in the execution JSON.
route (io.kestra.plugin.core.flow.If): zero duplicate keys logs clean and stops. One or more sends alert to Slack and 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. hold (io.kestra.plugin.core.flow.Pause) waits for an answer.
On approval remediate (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) walks the surplus id batches and purge (io.kestra.plugin.mongodb.Delete, DELETE_MANY) removes each batch by _id. With nothing targeted the run logs nothing_to_delete.
reverify (io.kestra.plugin.mongodb.Aggregate) groups the collection by the key again and verify_gate fails the run with io.kestra.plugin.core.execution.Fail if any key still has more than one document. 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 business keys carried by more than one document and the total surplus, which no MongoDB command reports for you when the collection has no unique index.
A Slack alert the moment duplicates exist, before a report that trusts the key starts double counting.
A deterministic survivor per key, kept by a rule you pick and recorded in the log, so the copy that stays is a decision rather than an accident.
A gated, approved delete of only the surplus copies, by _id, with the survivor safe by construction.
A re-group that fails the run when any key is still not unique, so a partial clear or a writer still inserting copies is caught rather than hidden.

Who it's for

Teams on an at-least-once ingest or an upsert path that cannot add a unique index, so the same record arrives twice.
Data engineers whose counts drifted because a join on a supposedly unique key fans out.
Anyone preparing a collection for a unique index who must clear the existing duplicates first, because createIndex with unique: true fails while any duplicate is present.

Why orchestrate this with Kestra

A unique index is the right long-term fix, but it cannot be created through the bundled plugin and it fails outright until the existing duplicates are gone. The usual answer is a one-off script somebody runs by hand. This flow makes it a scheduled, reviewable job. The detection is one server-side pipeline, the survivor rule is explicit, the delete sits behind auto_remediate, dry_run and a Pause with the decision recorded on the execution, the batches are serialised by concurrencyLimit: 1 and the re-group turns the delete into a checked outcome. 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.orders was seeded with three documents sharing email of amy@example.com, two sharing ben@example.com, plus three singleton documents with their own emails. Every before and after number was read back with mongosh, not only from the Kestra log.

Run	Inputs	Result
Clean	`business_key: email` on a collection with no duplicates	0 duplicate keys. `clean` logged, no Slack alert, SUCCESS. `mongosh` confirms one document per email
Detect	`business_key: email`, defaults	2 duplicate keys, 3 surplus copies. Slack alert, SUCCESS. `mongosh` groups confirm amy 3, ben 2
Dry run	the same	Reports the delete it would make, nothing deleted. `mongosh` count unchanged
Armed but dry	`auto_remediate: true`, `dry_run: true`	Gate stayed closed, no `Pause`, nothing deleted
Declined	`auto_remediate: true`, `dry_run: false`, resumed `approved: false`	Paused, then `declined`. `mongosh` count unchanged
Remediate	the same, resumed `approved: true`	3 surplus deleted, re-group 0, SUCCESS. `mongosh` amy 1, ben 1, the three singletons untouched
Survivor by id	`survivor_strategy: oldest_id` vs `newest_id` on the same seed	`mongosh` confirms the kept `_id` is the min and the max respectively
Survivor by time	`survivor_strategy: newest_timestamp`	`mongosh` confirms the kept document is the one with the largest `createdAt`
Still dirty	remediate with a writer inserting another `amy@example.com` during the `Pause`	Surplus deleted, re-group still found a duplicate key, `verify_gate` failed the run, `errors` alerted
Unreachable	`connection_uri` pointed at a dead host	`detect` failed, `errors` posted to Slack, execution FAILED

Does the detector actually fail on duplicates

The clean run and the detect run isolate it. Nothing changed between them except the data: a collection with one document per email returned 0 duplicate keys, the seeded one returned 2 with 3 surplus copies. mongosh confirmed both with an aggregation grouping on email. The payoff is the other half: after the remediate run a reader grouping on email returned exactly one document per key, where before it saw three for amy@example.com. The still-dirty run proved the gate the other way, failing when a key stayed duplicated.

Prerequisites

A reachable MongoDB instance and a user that can read the collection and run a delete on it.
A business key that should identify one logical record. The _id of each document is expected to be a stock ObjectId, which the survivor rule and the delete round-trip rely on.
Local testing: docker run -d --name mdb --network mynet mongo:7.0, seed a few documents that share a key plus some singletons, 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`, `collection`	`appdata`, `orders`	The collection to deduplicate
`business_key`	`email`	The field or comma separated fields that identify one record
`survivor_strategy`	`oldest_id`	Which copy to keep: `oldest_id`, `newest_id` or `newest_timestamp`
`timestamp_field`	`createdAt`	The date field for `newest_timestamp`, ignored otherwise
`scan_limit`	`25`	How many duplicate keys the report names. Counts stay exact
`delete_limit`	`50000`	Ceiling on surplus copies deleted per run
`batch_size`	`500`	Surplus 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, collection and business_key, then run the flow. It reports the duplicate keys and the surplus. It deletes nothing.
Pick a survivor_strategy, set auto_remediate: true and dry_run: false, run it again, then answer the pause. The surplus is deleted and the re-group proves each key is unique.
Create the unique index the flow stands in for, so the key stays unique between runs.
Enable the daily schedule, which always runs in safe mode.

How to extend

Deduplicate several collections by wrapping the tasks in a io.kestra.plugin.core.flow.Loop over a list of collection and key pairs, with concurrencyLimit: 1 on the part that deletes.
Archive the surplus before deleting it: add a $merge of the surplus documents into a backup collection ahead of the delete, so a wrong survivor rule is recoverable.
Keep an index on the business key, so the detection group is served without a collection scan.
Replace the Pause with a Slack approval. Or drop it for a collection you are sure about.

MongoDB Duplicate Document Dedup 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 duplicates

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

MongoDB Duplicate Document Dedup 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 duplicates

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Meilisearch Indexing Queue Staleness Sentinel

MinIO Cluster Disk Healing Sentinel

Catch MinIO Replication Drift Before Objects Go Missing on the Target

Enforce a MinIO bucket lifecycle expiry and noncurrent version policy

Prune noncurrent object versions billed forever behind a versioned bucket

Scheduled MinIO Bucket Retention Cleanup

More Data Blueprints

Sync new users from Postgres to HubSpot

Huawei GeminiDB TTL Hygiene Sentinel

Apache Iceberg Partition Compaction and Lakehouse Maintenance Gate

Idempotent Incremental CSV Loader from S3 to PostgreSQL

Automate Apache Iceberg Table Maintenance and Snapshot Expiry via REST Catalog

Sync Hubspot CRM data to BigQuery on a schedule

New to Kestra?