Ingest a public JSON API into DocumentDB

id: documentdb-api-ingest
namespace: company.team

labels:
  team: data
  domain: ingestion

concurrency:
  limit: 1

inputs:
  - id: api_url
    type: STRING
    defaults: https://jsonplaceholder.typicode.com/users
    description: Public JSON API returning an array of records to ingest.

  - id: business_key
    type: STRING
    defaults: id
    description: |
      Field in each source record that uniquely identifies it. Used as the `external_id` upsert key
      so re-runs upsert in place instead of accumulating duplicates.

variables:
  database: app_data
  collection: customers

pluginDefaults:
  - type: io.kestra.plugin.documentdb
    values:
      host: "{{ secret('DOCUMENTDB_HOST') }}"
      database: "{{ vars.database }}"
      username: "{{ secret('DOCUMENTDB_USERNAME') }}"
      password: "{{ secret('DOCUMENTDB_PASSWORD') }}"

tasks:
  - id: fetch_api
    type: io.kestra.plugin.core.http.Request
    description: Fetch a JSON array of records from the public API.
    uri: "{{ inputs.api_url }}"
    method: GET
    contentType: application/json
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3

  - id: guard_array
    type: io.kestra.plugin.core.execution.Fail
    description: Fail fast if the body is empty or not a JSON array, before any write.
    condition: "{{ (fromJson(outputs.fetch_api.body) is iterable) == false or
      fromJson(outputs.fetch_api.body) | length == 0 }}"
    errorMessage: "Source {{ inputs.api_url }} did not return a non-empty JSON array."

  - id: ingest_records
    type: io.kestra.plugin.core.flow.ForEach
    description: Upsert every record in the feed, not just the first batch, keyed on
      a deterministic external_id.
    concurrencyLimit: 1
    values: "{{ fromJson(outputs.fetch_api.body) }}"
    tasks:
      - id: read_existing
        type: io.kestra.plugin.documentdb.Read
        description: Check whether a document with this external_id already exists,
          since Update has no native upsert flag.
        collection: "{{ vars.collection }}"
        filter:
          external_id: "{{ fromJson(taskrun.value)[inputs.business_key] }}"
        fetchType: FETCH_ONE
        retry:
          type: constant
          interval: PT10S
          maxAttempt: 3

      - id: upsert_record
        type: io.kestra.plugin.core.flow.If
        description: Update the matched document in place, otherwise insert it (emulated
          upsert).
        condition: "{{ outputs.read_existing.row is defined and
          outputs.read_existing.row is not null }}"
        then:
          - id: update_record
            type: io.kestra.plugin.documentdb.Update
            description: Refresh the existing document so re-runs and backfills overwrite in
              place instead of duplicating.
            collection: "{{ vars.collection }}"
            filter:
              external_id: "{{ fromJson(parent.taskrun.value)[inputs.business_key] }}"
            update:
              $set:
                record: "{{ parent.taskrun.value }}"
                source: "{{ inputs.api_url }}"
                last_ingested_at: "{{ trigger.date ?? execution.startDate }}"
              $inc:
                ingest_count: 1
            updateMany: false
            retry:
              type: constant
              interval: PT10S
              maxAttempt: 3
        else:
          - id: insert_record
            type: io.kestra.plugin.documentdb.Insert
            description: Insert the record the first time it is seen.
            collection: "{{ vars.collection }}"
            document:
              external_id: "{{ fromJson(parent.taskrun.value)[inputs.business_key] }}"
              record: "{{ parent.taskrun.value }}"
              source: "{{ inputs.api_url }}"
              last_ingested_at: "{{ trigger.date ?? execution.startDate }}"
              ingest_count: 1
            retry:
              type: constant
              interval: PT10S
              maxAttempt: 3

  - id: count_ingested
    type: io.kestra.plugin.documentdb.Read
    description: Count the records present in the collection after the load for the
      run summary.
    collection: "{{ vars.collection }}"
    fetchType: STORE
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3

  - id: insert_run_marker
    type: io.kestra.plugin.documentdb.Insert
    description: Record a schedule-aware ingestion audit marker after a successful load.
    collection: ingestion_runs
    document:
      source: "{{ inputs.api_url }}"
      source_record_count: "{{ fromJson(outputs.fetch_api.body) | length }}"
      collection_size_after: "{{ outputs.count_ingested.size }}"
      status: success
      execution_id: "{{ execution.id }}"
      schedule_date: "{{ trigger.date ?? execution.startDate }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3

  - id: log_summary
    type: io.kestra.plugin.core.log.Log
    message: "Ingested {{ fromJson(outputs.fetch_api.body) | length }} source
      records from {{ inputs.api_url }}; {{ vars.collection }} now holds {{
      outputs.count_ingested.size }} documents."

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 }})."

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