Klaviyo Send Job Monitoring with Slack Alerts

id: klaviyo-send-job-monitoring
namespace: company.marketing

labels:
  team: marketing-ops
  monitor: klaviyo-send

inputs:
  - id: job_ids
    type: ARRAY
    itemType: STRING
    description: |
      Klaviyo campaign send job IDs to monitor. Best used for jobs that are
      still in flight; once a job reaches a terminal state it stops changing,
      so prune completed IDs from this list (or feed it from a discovery step
      that lists recently created send jobs).
    defaults:
      - "01HZZZZZZZZZZZZZZZZZZZZZZZ"
  - id: alerting_statuses
    type: ARRAY
    itemType: STRING
    description: |
      Send job statuses that should page the team. Verify these against the
      Klaviyo Campaign Send Job API for your account before relying on them.
      "queued" is intentionally NOT treated as healthy here; a job stuck in a
      pre-send state is exactly what monitoring should surface.
    defaults:
      - cancelled
      - failed
      - error

pluginDefaults:
  - type: io.kestra.plugin.klaviyo
    values:
      apiKey: "{{ secret('KLAVIYO_API_KEY') }}"
  - type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    values:
      url: "{{ secret('SLACK_WEBHOOK_URL') }}"

tasks:
  - id: get_send_jobs
    type: io.kestra.plugin.klaviyo.jobs.GetSendJob
    description: |
      Fetch the current state of every send job in the input. fetchType FETCH
      returns all rows so each job is evaluated, not just the first.
    jobIds: "{{ inputs.job_ids }}"
    fetchType: FETCH
    retry:
      type: constant
      interval: PT30S
      maxAttempt: 4
    timeout: PT5M

  - id: evaluate_jobs
    type: io.kestra.plugin.core.flow.ForEach
    description: Evaluate each fetched send job and alert only on a transition into
      a bad state.
    concurrencyLimit: 1
    values: "{{ outputs.get_send_jobs.rows ?? [] }}"
    tasks:
      - id: get_last_status
        type: io.kestra.plugin.core.kv.Get
        description: Read the last status we recorded for this job so we only alert on
          change.
        key: "klaviyo_send_job_{{ fromJson(taskrun.value).id }}"
        errorOnMissing: false

      - id: route_status
        type: io.kestra.plugin.core.flow.If
        description: |
          Alert only when the job is in an alerting status AND that differs
          from the last status we saw, so a failed job pages once on transition
          rather than every poll.
        condition: "{{ (inputs.alerting_statuses contains
          fromJson(taskrun.value).attributes.status) and
          (fromJson(taskrun.value).attributes.status !=
          (outputs.get_last_status.value ?? '')) }}"
        then:
          - id: alert_unhealthy_send
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Page the marketing ops channel about a send job that just entered a
              bad state.
            payload: |
              {
                "blocks": [
                  {
                    "type": "header",
                    "text": {
                      "type": "plain_text",
                      "text": "Klaviyo send job needs attention"
                    }
                  },
                  {
                    "type": "section",
                    "fields": [
                      {
                        "type": "mrkdwn",
                        "text": "*Send job status:*\n{{ fromJson(taskrun.value).attributes.status }}"
                      },
                      {
                        "type": "mrkdwn",
                        "text": "*Send job id:*\n{{ fromJson(taskrun.value).id }}"
                      },
                      {
                        "type": "mrkdwn",
                        "text": "*Previous status:*\n{{ outputs.get_last_status.value ?? 'none' }}"
                      },
                      {
                        "type": "mrkdwn",
                        "text": "*Execution:*\n{{ execution.id }}"
                      }
                    ]
                  },
                  {
                    "type": "context",
                    "elements": [
                      {
                        "type": "mrkdwn",
                        "text": "Open Klaviyo to inspect the send job and resend or reschedule as needed."
                      }
                    ]
                  }
                ]
              }
        else:
          - id: log_status
            type: io.kestra.plugin.core.log.Log
            description: Record the current status for the audit trail without paging.
            message: "Send job {{ fromJson(taskrun.value).id }} status is {{
              fromJson(taskrun.value).attributes.status }} (previous {{
              outputs.get_last_status.value ?? 'none' }}); no new alert."

      - id: record_status
        type: io.kestra.plugin.core.kv.Set
        description: |
          Persist the current status as the last-seen value so the next poll
          only alerts on a further transition. Written after the alert branch
          so a failed Slack post does not suppress a future page.
        key: "klaviyo_send_job_{{ fromJson(taskrun.value).id }}"
        value: "{{ fromJson(taskrun.value).attributes.status }}"
        kvType: STRING
        ttl: P30D

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert if the monitoring flow itself fails (for example a bad API key).
    messageText: ":rotating_light: {{ flow.namespace }}.{{ flow.id }} monitoring
      failed (execution {{ execution.id }})."

triggers:
  - id: business_hours
    type: io.kestra.plugin.core.trigger.Schedule
    description: Poll the send jobs every 30 minutes during business hours (08:00 to
      18:00 UTC, Mon to Fri).
    cron: "*/30 8-18 * * 1-5"
    timezone: Etc/UTC