Promote SQLMesh changes to production only when data audits pass

id: sqlmesh-audit-gated-promotion
namespace: company.team

inputs:
  - id: project_git_url
    type: STRING
    displayName: SQLMesh project Git URL
    description: Git repository that holds the SQLMesh project. The default is the
      public Tobiko sushi example, so the flow runs with no setup.
    defaults: https://github.com/TobikoData/sqlmesh-examples.git

  - id: project_branch
    type: STRING
    displayName: Git branch
    defaults: main

  - id: project_subdir
    type: STRING
    displayName: Project subdirectory
    description: Path inside the repository that holds the SQLMesh config.yaml.
    defaults: 001_sushi/1_simple

  - id: max_waiter_daily_revenue
    type: INT
    displayName: Max revenue per waiter per day
    description: >
      Threshold used by the revenue range audit. The sushi data peaks at about
      570, so the default passes. Set it to 500 to watch the gate block the
      release.
    defaults: 1000

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Post the result to Slack. Needs the SLACK_WEBHOOK_URL secret.
    defaults: false

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    runIf: '{{ inputs.notify_slack }}'
    description: A failed build, audit or promotion must not pass silently.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'SQLMesh audit-gated promotion failed (execution ' ~ execution.id ~ '). Check the execution logs: production was not promoted unless the promote_production task succeeded.'} | toJson }}

triggers:
  - id: on_merge
    type: io.kestra.plugin.core.trigger.Webhook
    description: >
      Shipped disabled. Call this webhook from CI after a merge to main to gate
      every change. Replace the key with a long random value, then enable it.
    key: replace-with-a-long-random-key
    disabled: true

tasks:
  - id: stage_and_audit
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One working directory, so the clone, the staging build and the
      audit share one checkout.
    tasks:
      - id: clone
        type: io.kestra.plugin.git.Clone
        description: Check out the SQLMesh project at the branch under test.
        url: "{{ inputs.project_git_url }}"
        branch: "{{ inputs.project_branch }}"
        depth: 1

      - id: plan_staging
        type: io.kestra.plugin.sqlmesh.cli.SQLMeshCLI
        description: >
          Add the data contract as SQLMesh standalone audits, then build every
          change in the isolated dev environment. Production is not touched.
        inputFiles:
          "{{ inputs.project_subdir }}/audits/assert_top_waiters_have_names.sql": |
            AUDIT (
              name assert_top_waiters_have_names,
              standalone true,
              depends_on [sushisimple.top_waiters]
            );
            SELECT * FROM sushisimple.top_waiters WHERE waiter_name IS NULL
          "{{ inputs.project_subdir }}/audits/assert_waiter_daily_revenue_in_range.sql": |
            AUDIT (
              name assert_waiter_daily_revenue_in_range,
              standalone true,
              depends_on [sushisimple.waiter_revenue_by_day]
            );
            SELECT * FROM sushisimple.waiter_revenue_by_day
            WHERE revenue < 0 OR revenue > {{ inputs.max_waiter_daily_revenue }}
        env:
          PROJECT_SUBDIR: "{{ inputs.project_subdir }}"
        beforeCommands:
          - cd "$PROJECT_SUBDIR"
        commands:
          - sqlmesh plan dev --no-prompts --auto-apply

      - id: audit
        type: io.kestra.plugin.sqlmesh.cli.SQLMeshCLI
        description: >
          Run every audit against the staging build. The task records passed or
          failed plus the names of the failing audits. It does not fail itself,
          so the gate below decides. It also exports the project with its DuckDB
          state, so promotion uses this exact audited build.
        env:
          PROJECT_SUBDIR: "{{ inputs.project_subdir }}"
        beforeCommands:
          - cd "$PROJECT_SUBDIR"
        commands:
          - |
            if sqlmesh audit > audit.log 2>&1; then status=passed; else status=failed; fi
            cat audit.log
            failed=$(grep -o 'Failure in audit [A-Za-z0-9_]*' audit.log | awk '{print $4}' | sort -u | paste -sd ',' -)
            if [ "$status" = failed ] && [ -z "$failed" ]; then failed="none parsed, see the audit task log"; fi
            echo "::{\"outputs\":{\"audit_status\":\"$status\",\"failed_audits\":\"$failed\"}}::"
        outputFiles:
          - "{{ inputs.project_subdir }}/**"

  - id: promotion_gate
    type: io.kestra.plugin.core.flow.If
    description: Promote to production only when every audit passed.
    condition: "{{ outputs.audit.vars.audit_status == 'passed' }}"
    then:
      - id: promote_production
        type: io.kestra.plugin.sqlmesh.cli.SQLMeshCLI
        description: >
          Restore the audited project and state, then promote it to prod.
          SQLMesh reuses the tables built in dev, so this is a virtual layer
          swap and not a rebuild. The last command prints the promoted result.
        inputFiles: "{{ outputs.audit.outputFiles }}"
        env:
          PROJECT_SUBDIR: "{{ inputs.project_subdir }}"
        beforeCommands:
          - cd "$PROJECT_SUBDIR"
        commands:
          - sqlmesh plan prod --no-prompts --auto-apply
          - sqlmesh fetchdf "SELECT * FROM sushisimple.top_waiters ORDER BY
            revenue DESC"
        outputFiles:
          - "{{ inputs.project_subdir }}/db/*.db"

      - id: notify_promoted
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack_promoted
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "SQLMesh: all audits passed in dev, promoted to prod. Execution {{ execution.id }}."
              }
    else:
      - id: notify_blocked
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack_blocked
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "SQLMesh: promotion to prod BLOCKED. Failing audits: {{ outputs.audit.vars.failed_audits }}. Execution {{ execution.id }}."
              }

      - id: block_release
        type: io.kestra.plugin.core.execution.Fail
        description: End the run as FAILED, so a blocked release shows red in Kestra and
          in the CI job that called the webhook.
        errorMessage: "Promotion blocked. Failing audits: {{
          outputs.audit.vars.failed_audits }}"