DocumentDB analytics report with Slack digest

id: documentdb-read-transform-report
namespace: company.analytics

labels:
  team: analytics
  domain: reporting

variables:
  database: app_data
  collection: orders

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

tasks:
  - id: aggregate_orders
    type: io.kestra.plugin.documentdb.Read
    description: Group the scheduled day's orders by status and compute counts and
      revenue, stored as Ion.
    collection: "{{ vars.collection }}"
    aggregationPipeline:
      - $match:
          created_at:
            $gte:
              $date: "{{ (trigger.date ?? execution.startDate) | dateAdd(-1, 'DAYS') |
                date('yyyy-MM-dd') }}T00:00:00Z"
            $lt:
              $date: "{{ (trigger.date ?? execution.startDate) | date('yyyy-MM-dd')
                }}T00:00:00Z"
      - $group:
          _id: "$status"
          orders:
            $sum: 1
          revenue:
            $sum: "$amount"
      - $sort:
          revenue: -1
    fetchType: STORE
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3

  - id: to_csv
    type: io.kestra.plugin.serdes.csv.IonToCsv
    description: Convert the stored Ion aggregation result into a downloadable CSV report.
    from: "{{ outputs.aggregate_orders.uri }}"
    header: true

  - id: fetch_groups
    type: io.kestra.plugin.documentdb.Read
    description: Re-read the same aggregation as in-memory rows so the digest can
      surface real revenue figures.
    collection: "{{ vars.collection }}"
    aggregationPipeline:
      - $match:
          created_at:
            $gte:
              $date: "{{ (trigger.date ?? execution.startDate) | dateAdd(-1, 'DAYS') |
                date('yyyy-MM-dd') }}T00:00:00Z"
            $lt:
              $date: "{{ (trigger.date ?? execution.startDate) | date('yyyy-MM-dd')
                }}T00:00:00Z"
      - $group:
          _id: "$status"
          orders:
            $sum: 1
          revenue:
            $sum: "$amount"
      - $sort:
          revenue: -1
    fetchType: FETCH
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3

  - id: report
    type: io.kestra.plugin.core.flow.If
    description: Post the digest with real figures, or a distinct no-orders message
      when the window is empty.
    condition: "{{ (outputs.fetch_groups.size ?? 0) > 0 }}"
    then:
      - id: notify_digest
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post a daily orders digest with per-status order and revenue figures.
        url: "{{ secret('SLACK_WEBHOOK') }}"
        retry:
          type: constant
          interval: PT10S
          maxAttempt: 3
        payload: |
          {
            "blocks": [
              {
                "type": "header",
                "text": {
                  "type": "plain_text",
                  "text": "Daily orders digest"
                }
              },
              {
                "type": "section",
                "fields": [
                  {
                    "type": "mrkdwn",
                    "text": "*Report day:*\n{{ (trigger.date ?? execution.startDate) | dateAdd(-1, 'DAYS') | date('yyyy-MM-dd') }}"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Status groups:*\n{{ outputs.fetch_groups.size }}"
                  }
                ]
              },
              {
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "{% for g in outputs.fetch_groups.rows %}*{{ g._id }}*: {{ g.orders }} orders, revenue {{ g.revenue }}\n{% endfor %}"
                }
              },
              {
                "type": "context",
                "elements": [
                  {
                    "type": "mrkdwn",
                    "text": "Flow {{ flow.namespace }}.{{ flow.id }} | execution {{ execution.id }} | CSV {{ outputs.to_csv.uri }}"
                  }
                ]
              }
            ]
          }
    else:
      - id: notify_empty
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post a distinct message so an empty window is never confused with a
          broken pipeline.
        url: "{{ secret('SLACK_WEBHOOK') }}"
        retry:
          type: constant
          interval: PT10S
          maxAttempt: 3
        messageText: ":information_source: Daily orders digest for {{ (trigger.date ??
          execution.startDate) | dateAdd(-1, 'DAYS') | date('yyyy-MM-dd') }}: no
          orders in the window. Flow {{ flow.namespace }}.{{ flow.id }}
          (execution {{ execution.id }})."

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 daily orders digest did not run."

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