DocumentDB Upsert and Sync Blueprint for Kestra

id: documentdb-upsert-sync
namespace: company.team

triggers:
  - id: hourly_sync
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"
    timezone: Etc/UTC

labels:
  team: data
  domain: sync

concurrency:
  limit: 1

inputs:
  - id: source_url
    type: STRING
    defaults: https://jsonplaceholder.typicode.com/users/1
    description: External source returning the current record to sync.

  - id: retire_when_gone
    type: BOOL
    defaults: false
    description: |
      When true and the source returns 404/410 (record retired at the source), delete exactly the one
      matching document. Defaults to false. There is no collection-wide age-based prune: a single-record
      sync cannot safely decide that other documents are stale, so retention is opt-in and id-scoped only.

variables:
  database: app_data
  collection: catalog

tasks:
  - id: fetch_source
    type: io.kestra.plugin.core.http.Request
    description: Pull the current record from the external source of truth (404/410
      means retired).
    uri: "{{ inputs.source_url }}"
    method: GET
    contentType: application/json
    options:
      allowFailed: true
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3

  - id: route
    type: io.kestra.plugin.core.flow.If
    description: Branch on whether the source returned a live record or signalled
      the record is retired.
    condition: "{{ outputs.fetch_source.code == 200 }}"
    then:
      - id: guard_body
        type: io.kestra.plugin.core.execution.Fail
        description: Validate the source body has the scalar fields the sync needs
          before any write.
        condition: "{{ (fromJson(outputs.fetch_source.body).id is defined and
          fromJson(outputs.fetch_source.body).name is defined and
          fromJson(outputs.fetch_source.body).email is defined) == false }}"
        errorMessage: "Source body from {{ inputs.source_url }} is missing required
          scalar fields id/name/email."

      - id: read_existing
        type: io.kestra.plugin.documentdb.Read
        description: Check whether a document with this external_id already exists.
        host: "{{ secret('DOCUMENTDB_HOST') }}"
        database: "{{ vars.database }}"
        username: "{{ secret('DOCUMENTDB_USERNAME') }}"
        password: "{{ secret('DOCUMENTDB_PASSWORD') }}"
        collection: "{{ vars.collection }}"
        filter:
          external_id: "{{ fromJson(outputs.fetch_source.body).id }}"
        fetchType: FETCH_ONE
        retry:
          type: constant
          interval: PT10S
          maxAttempts: 3

      - id: upsert
        type: io.kestra.plugin.core.flow.If
        description: Update when the record exists, otherwise insert (emulated upsert;
          Update has no native upsert flag).
        condition: "{{ outputs.read_existing.row is defined and
          outputs.read_existing.row is not null }}"
        then:
          - id: update_existing
            type: io.kestra.plugin.documentdb.Update
            description: Refresh the matched document and bump the sync counter.
            host: "{{ secret('DOCUMENTDB_HOST') }}"
            database: "{{ vars.database }}"
            username: "{{ secret('DOCUMENTDB_USERNAME') }}"
            password: "{{ secret('DOCUMENTDB_PASSWORD') }}"
            collection: "{{ vars.collection }}"
            filter:
              external_id: "{{ fromJson(outputs.fetch_source.body).id }}"
            update:
              $set:
                name: "{{ fromJson(outputs.fetch_source.body).name }}"
                email: "{{ fromJson(outputs.fetch_source.body).email }}"
                synced_at: "{{ trigger.date ?? execution.startDate }}"
              $inc:
                sync_count: 1
            updateMany: false
            retry:
              type: constant
              interval: PT10S
              maxAttempts: 3
        else:
          - id: insert_new
            type: io.kestra.plugin.documentdb.Insert
            description: Insert the record for the first time when no match exists.
            host: "{{ secret('DOCUMENTDB_HOST') }}"
            database: "{{ vars.database }}"
            username: "{{ secret('DOCUMENTDB_USERNAME') }}"
            password: "{{ secret('DOCUMENTDB_PASSWORD') }}"
            collection: "{{ vars.collection }}"
            document:
              external_id: "{{ fromJson(outputs.fetch_source.body).id }}"
              name: "{{ fromJson(outputs.fetch_source.body).name }}"
              email: "{{ fromJson(outputs.fetch_source.body).email }}"
              sync_count: 1
              synced_at: "{{ trigger.date ?? execution.startDate }}"
            retry:
              type: constant
              interval: PT10S
              maxAttempts: 3

      - id: log_synced
        type: io.kestra.plugin.core.log.Log
        message: "Synced external_id={{ fromJson(outputs.fetch_source.body).id }}: {{
          outputs.upsert.evaluationResult ? 'updated existing' : 'inserted new'
          }}."
    else:
      - id: handle_retired
        type: io.kestra.plugin.core.flow.If
        description: Optionally delete the single retired document, scoped to its
          external_id, only when opted in.
        condition: "{{ inputs.retire_when_gone and (outputs.fetch_source.code == 404 or
          outputs.fetch_source.code == 410) }}"
        then:
          - id: delete_retired
            type: io.kestra.plugin.documentdb.Delete
            description: Delete exactly the one document whose source record is retired
              (scoped, never collection-wide).
            host: "{{ secret('DOCUMENTDB_HOST') }}"
            database: "{{ vars.database }}"
            username: "{{ secret('DOCUMENTDB_USERNAME') }}"
            password: "{{ secret('DOCUMENTDB_PASSWORD') }}"
            collection: "{{ vars.collection }}"
            filter:
              external_id: "{{ inputs.source_url | split('/') | last }}"
            deleteMany: false
            retry:
              type: constant
              interval: PT10S
              maxAttempts: 3

          - id: log_retired
            type: io.kestra.plugin.core.log.Log
            message: "Source returned {{ outputs.fetch_source.code }} for {{
              inputs.source_url }}; removed {{
              outputs.delete_retired.deletedCount }} retired document(s)."
        else:
          - id: log_skipped
            type: io.kestra.plugin.core.log.Log
            message: "Source returned {{ outputs.fetch_source.code }} for {{
              inputs.source_url }}; no write performed (retire_when_gone={{
              inputs.retire_when_gone }})."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    messageText: ":rotating_light: {{ flow.namespace }}.{{ flow.id }} failed
      (execution {{ execution.id }}). The collection may be partially synced;
      check before re-running."

Keep a DocumentDB collection continuously in sync with an external HTTP source of truth, one record per run, using a race-safe emulated upsert. The DocumentDB plugin's Update task has no native upsert flag, so this blueprint reads by external_id, then conditionally updates or inserts. It solves the classic time-of-check / time-of-use race in read-then-write upserts by serializing executions, and it avoids the silent data loss of age-based collection prunes by making record retirement strictly opt-in and scoped to a single document. Use it for change-data-capture style mirroring, reference-data sync, and lightweight CDC into Amazon DocumentDB or any MongoDB-compatible Data REST API.

How it works

fetch_source (io.kestra.plugin.core.http.Request) pulls the current record over HTTP with retry and options.allowFailed: true, so a 404 or 410 routes to the retirement branch instead of failing the run.
route (io.kestra.plugin.core.flow.If) branches on outputs.fetch_source.code == 200.
On a live record, guard_body (io.kestra.plugin.core.execution.Fail) validates the body has scalar id, name, and email before any write; read_existing (io.kestra.plugin.documentdb.Read, fetchType: FETCH_ONE) looks up the document by external_id.
upsert (io.kestra.plugin.core.flow.If) chooses update_existing (io.kestra.plugin.documentdb.Update, updateMany: false, with $set and a $inc on sync_count) when the row exists, otherwise insert_new (io.kestra.plugin.documentdb.Insert).
When the source returns 404 or 410 and retire_when_gone is true, delete_retired (io.kestra.plugin.documentdb.Delete, deleteMany: false) removes exactly that one document; otherwise the run logs and writes nothing.
The hourly_sync Schedule trigger fires every hour, and concurrency.limit: 1 serializes runs so no two executions interleave.

What you get

A safe upsert that never double-inserts under concurrency.
Opt-in, id-scoped retirement instead of dangerous collection-wide deletes.
Source-body validation before any database write.
A Slack failure alert carrying the execution ID so a partial sync is never silent.
A sync_count counter and synced_at timestamp on every document.

Who it's for

Data engineers mirroring an external API into DocumentDB.
Platform teams running reference-data or master-data sync.
Anyone needing dependable CDC-style upserts without a heavy ETL stack.

Why orchestrate this with Kestra

The DocumentDB API has no scheduler, no retries, and no notion of a serialized read-then-write across runs. Kestra adds the event-driven Schedule trigger, per-task retry blocks, concurrency.limit: 1 to close the upsert race, full execution lineage and outputs for audit, and an errors block for alerting, all in declarative YAML you can version and review. Pair it with a unique index on external_id so even an external writer cannot create a duplicate.

Prerequisites

A reachable DocumentDB Data REST API endpoint with a catalog collection keyed by external_id (unique index recommended).
An external HTTP source returning JSON with scalar id, name, and email, and 404/410 when a record is retired.
Outbound network access from Kestra workers to both the source and DocumentDB.

Secrets

DOCUMENTDB_HOST: base HTTPS endpoint of the DocumentDB Data REST API.
DOCUMENTDB_USERNAME: basic-auth username for the DocumentDB API.
DOCUMENTDB_PASSWORD: basic-auth password for the DocumentDB API.
SLACK_WEBHOOK: Slack incoming webhook URL used by the failure alert.

Quick start

Add the four secrets above to your Kestra instance.
Point the source_url input at your record endpoint and set the database and collection variables.
Run once to confirm the insert path, then run again to confirm the update path (collection size unchanged, sync_count incremented).
Enable the hourly schedule, and turn on retire_when_gone only once you trust the source to return 404/410 for genuinely retired records.

How to extend

Swap the single-record fetch for a paged endpoint and loop with Loop to sync many records per run.
Replace the $set fields to mirror your real document shape.
Add a downstream notification or a metrics task keyed on outputs.upsert.evaluationResult.
Trigger on a webhook instead of a schedule for near-real-time sync.

Upsert a DocumentDB record from an external source

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Upsert a DocumentDB record from an external source

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Kafka Consumer Group Offset Checkpoint and Rewind

Kafka Consumer Lag Sentinel

Validate and Reshape a Webhook Order Payload with JSONata

Kafka Client Quota Guard

Fetch JSON from an API in Python and load it into MongoDB

Kafka DLQ Quarantine, Classification, and Safe Replay Gate

New to Kestra?