Data Quality Incident Escalation with Soda

id: soda-incident-escalation
namespace: company.team
description: |
  Treat a failed data quality check as a production incident. A Soda
  scan runs with allowFailure so an If task can inspect the hasFailures
  output; failures open a Squadcast incident through its incoming
  webhook and notify Slack with the incident context, while clean scans
  log quietly. Bad data pages someone, the same as a down service.

inputs:
  - id: warehouse_path
    type: STRING
    defaults: /data/warehouse.duckdb
    description: Persistent path for the DuckDB warehouse file. Point at a volume
      that survives task runs so the scan sees the same file the load wrote.

tasks:
  - id: load_transactions
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Stand-in for the real load step. Creates and refreshes a small
      transactions table so the scan has data to judge; replace with your
      ingestion task.
    url: "jdbc:duckdb:{{ inputs.warehouse_path }}"
    sql: |
      CREATE TABLE IF NOT EXISTS transactions (
          transaction_id INTEGER,
          account_id VARCHAR,
          amount DOUBLE,
          created_at TIMESTAMP
      );
      DELETE FROM transactions;
      INSERT INTO transactions VALUES
          (1, 'acc-001', 120.00, now()),
          (2, 'acc-002', 75.40, now()),
          (3, 'acc-003', 310.25, now());
    fetchType: NONE

  - id: incident_scan
    type: io.kestra.plugin.soda.Scan
    description: Execute SodaCL checks on the transactions table. allowFailure keeps
      a failed check from ending the flow before the escalation branch runs,
      since paging is the point of this flow, not a side effect.
    allowFailure: true
    configuration:
      data_source kestra:
        type: duckdb
        path: "{{ inputs.warehouse_path }}"
    checks:
      checks for transactions:
        - row_count > 0
        - missing_count(account_id) = 0
        - duplicate_count(transaction_id) = 0
    requirements:
      - soda-core-duckdb

  - id: route_incident
    type: io.kestra.plugin.core.flow.If
    description: Decide whether this execution is an incident. Failed checks open a
      Squadcast incident and notify Slack; clean scans log a quiet all-clear.
    condition: "{{ outputs.incident_scan.result.hasFailures }}"
    then:
      - id: page_incident
        type: io.kestra.plugin.squadcast.SquadcastIncomingWebhook
        description: Open a Squadcast incident carrying the flow, table, and execution
          id, so on-call lands on the failing scan without hunting through
          channels.
        url: "{{ secret('SQUADCAST_WEBHOOK_URL') }}"
        payload: |
          {
            "message": "Data quality incident in {{ flow.namespace }}.{{ flow.id }}",
            "description": "Soda checks failed on the transactions table in execution {{ execution.id }}. failures={{ outputs.incident_scan.result.hasFailures }}, warnings={{ outputs.incident_scan.result.hasWarnings }} across {{ outputs.incident_scan.result.checks | length }} checks. Inspect the incident_scan task logs for the failing check.",
            "status": "trigger",
            "event_id": "{{ execution.id }}",
            "tags": {
              "flow": "{{ flow.namespace }}.{{ flow.id }}",
              "source": "soda-scan",
              "table": "transactions"
            }
          }
      - id: notify_oncall_channel
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Mirror the page into Slack so the wider team sees the incident
          context while on-call works it in Squadcast.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Squadcast incident opened for failing Soda checks on the transactions table in flow {{ flow.id }} (execution {{ execution.id }}). On-call has been paged; treat the table as unreliable until the incident resolves."
          }
    else:
      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        description: Record the all-clear so the execution history shows every scan that
          judged the table healthy.
        message: "No incident: {{ outputs.incident_scan.result.checks | length }} Soda
          checks passed on the transactions table."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the escalation flow itself breaks, because a paging
      pipeline that fails silently means incidents stop being opened while
      everyone assumes they would be.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Soda incident escalation flow {{ flow.id }} FAILED to run (execution {{ execution.id }}). Quality incidents are currently not being opened; check the scan and webhook tasks."
      }

triggers:
  - id: after_load_window
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scan every morning after the nightly loads should have landed, so
      incidents open at the start of the working day.
    cron: "0 7 * * *"
    disabled: true