Open a Squadcast Incident on Data Quality Breach

id: squadcast-data-quality-incident
namespace: company.team
description: |
  Treat bad data as an incident, not an email. A Python quality gate
  computes the null rate of a key column over the latest batch and emits
  it through Kestra's output protocol. When the rate breaches the
  threshold, the flow opens a Squadcast incident carrying the numbers;
  clean batches log and pass through.

inputs:
  - id: max_null_rate
    type: FLOAT
    defaults: 0.1
    description: Maximum tolerated fraction of records with a null email before
      on-call is paged. The inline sample breaches this on purpose so the
      incident path is visible on first run.

tasks:
  - id: profile_orders
    type: io.kestra.plugin.scripts.python.Script
    description: Profile the latest batch, an inline sample standing in for your
      real load, and emit the null rate of the email column as task outputs.
    script: |
      import json

      records = [
          {"order_id": 101, "email": "ana@example.com"},
          {"order_id": 102, "email": "ben@example.com"},
          {"order_id": 103, "email": None},
          {"order_id": 104, "email": "cara@example.com"},
          {"order_id": 105, "email": None},
          {"order_id": 106, "email": "dan@example.com"},
          {"order_id": 107, "email": "eva@example.com"},
          {"order_id": 108, "email": None},
          {"order_id": 109, "email": "finn@example.com"},
          {"order_id": 110, "email": "gia@example.com"},
          {"order_id": 111, "email": "hugo@example.com"},
          {"order_id": 112, "email": "ines@example.com"}
      ]

      null_count = sum(1 for r in records if not r["email"])
      null_rate = round(null_count / len(records), 3)
      print("::" + json.dumps({"outputs": {"null_rate": null_rate, "null_count": null_count, "row_count": len(records)}}) + "::")

  - id: quality_gate
    type: io.kestra.plugin.core.flow.If
    description: Route the batch on the measured null rate, page on breach, log on a
      clean profile.
    condition: "{{ outputs.profile_orders.vars.null_rate > inputs.max_null_rate }}"
    then:
      - id: page_quality_incident
        type: io.kestra.plugin.squadcast.SquadcastIncomingWebhook
        description: Open a Squadcast incident with the measured rate, the threshold,
          and the affected row counts, so on-call can judge blast radius without
          rerunning the profile.
        url: "{{ secret('SQUADCAST_WEBHOOK_URL') }}"
        payload: |
          {
            "message": "Data quality breach in {{ flow.namespace }}.{{ flow.id }}",
            "description": "Null rate on orders.email is {{ outputs.profile_orders.vars.null_rate }} against a threshold of {{ inputs.max_null_rate }}. {{ outputs.profile_orders.vars.null_count }} of {{ outputs.profile_orders.vars.row_count }} records are affected in execution {{ execution.id }}.",
            "status": "trigger",
            "event_id": "{{ execution.id }}",
            "tags": {
              "flow": "{{ flow.namespace }}.{{ flow.id }}",
              "check": "null_rate",
              "column": "orders.email"
            }
          }
    else:
      - id: log_quality_ok
        type: io.kestra.plugin.core.log.Log
        description: Record the passing profile so quality history lives alongside the
          executions that produced it.
        message: "Quality gate passed, null rate {{
          outputs.profile_orders.vars.null_rate }} within threshold {{
          inputs.max_null_rate }}."

errors:
  - id: report_pager_failure
    type: io.kestra.plugin.discord.DiscordIncomingWebhook
    description: If the profile or the Squadcast call fails, the quality gate is
      blind. Report it to Discord so bad data cannot slip through while the
      pager is down.
    url: "{{ secret('DISCORD_WEBHOOK_URL') }}"
    payload: |
      {
        "content": "Data quality pager flow {{ flow.id }} FAILED (execution {{ execution.id }}). The batch was not profiled or the page did not fire; treat the latest data as unverified."
      }