Liquibase deployment on Git push with rehearsal, approval and auto-rollback

id: liquibase-migrate-on-push-with-rehearsal
namespace: company.platform

inputs:
  - id: commit
    type: STRING
    displayName: Commit
    description: SHA of the changelog repository commit to deploy. A push fills it
      in with the pushed commit.
    validator: ^[0-9a-f]{7,40}$

  - id: pushed_by
    type: STRING
    displayName: Pushed by
    required: false

variables:
  git_url: http://localhost:3000/platform/app-db.git
  git_username: kestra
  branch: main
  changelog_dir: db
  changelog_file: db/changelog.yaml
  verify_file: db/verify.sql
  prod_db: localhost:5432/app
  prod_user: app_migrator
  scratch_db: localhost:5432/app_rehearsal
  scratch_user: rehearsal
  approval_window: P1D
  liquibase_image: ghcr.io/kestra-io/liquibase:latest
  postgres_image: postgres:17-alpine

concurrency:
  limit: 1

tasks:
  - id: label_commit
    type: io.kestra.plugin.core.execution.Labels
    labels:
      commit: "{{ inputs.commit }}"
      pushed_by: "{{ (inputs.pushed_by ?? '') == '' ? 'manual' : inputs.pushed_by }}"

  - id: checkout
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      - id: clone
        type: io.kestra.plugin.git.Clone
        url: "{{ vars.git_url }}"
        username: "{{ vars.git_username }}"
        password: "{{ secret('GIT_TOKEN') }}"
        branch: "{{ vars.branch }}"
        commit: "{{ inputs.commit }}"

      - id: plan
        type: io.kestra.plugin.liquibase.CLI
        description: |
          `update-sql` writes the SQL that `update` would run on production, without running it. One `-- Changeset` comment per pending changeset gives the count. `changes.sql` is the same script without Liquibase's own lock and changelog statements, for the approver.
          The changelog files go to the execution's storage with the preview, so every later step works on exactly the files that were rehearsed and approved.
        containerImage: "{{ vars.liquibase_image }}"
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          networkMode: host
          user: root
        env:
          LIQUIBASE_COMMAND_URL: "jdbc:postgresql://{{ vars.prod_db }}"
          LIQUIBASE_COMMAND_USERNAME: "{{ vars.prod_user }}"
          LIQUIBASE_COMMAND_PASSWORD: "{{ secret('PROD_DB_PASSWORD') }}"
          LIQUIBASE_COMMAND_CHANGELOG_FILE: "{{ vars.changelog_file }}"
          LIQUIBASE_SHOW_BANNER: "false"
          LIQUIBASE_ANALYTICS_ENABLED: "false"
        outputFiles:
          - preview.sql
          - changes.sql
          - "{{ vars.changelog_dir }}/**"
        commands:
          - liquibase update-sql --output-file=preview.sql 2>&1
          - awk '/^-- Changeset /{print ""; print; next} /^--/ ||
            /databasechangelog/ || NF == 0 {next} {print}' preview.sql >
            changes.sql
          - echo "::{\"outputs\":{\"pending\":$(grep -c '^-- Changeset '
            preview.sql || true)}}::"

  - id: nothing_pending
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.plan.vars.pending == 0 }}"
    then:
      - id: label_up_to_date
        type: io.kestra.plugin.core.execution.Labels
        labels:
          outcome: up-to-date

      - id: log_up_to_date
        type: io.kestra.plugin.core.log.Log
        message: "{{ vars.prod_db }} already has every changeset of {{ inputs.commit }}.
          Nothing to deploy."

      - id: stop_up_to_date
        type: io.kestra.plugin.core.execution.Exit
        state: SUCCESS

  - id: rehearsal_copy
    type: io.kestra.plugin.scripts.shell.Commands
    description: Recreates the scratch database from the production schema and
      Liquibase's changelog table, so Liquibase sees the same pending changesets
      there. No business data is copied.
    containerImage: "{{ vars.postgres_image }}"
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: host
    env:
      PROD: "postgresql://{{ vars.prod_user }}@{{ vars.prod_db }}"
      PROD_PASSWORD: "{{ secret('PROD_DB_PASSWORD') }}"
      SCRATCH: "postgresql://{{ vars.scratch_user }}@{{ vars.scratch_db }}"
      SCRATCH_PASSWORD: "{{ secret('SCRATCH_DB_PASSWORD') }}"
      PGOPTIONS: -c client_min_messages=warning
    commands:
      - set -o pipefail
      - export PGPASSWORD="$SCRATCH_PASSWORD"
      - dropdb --if-exists --force --maintenance-db="${SCRATCH%/*}/postgres"
        "${SCRATCH##*/}"
      - createdb --maintenance-db="${SCRATCH%/*}/postgres" "${SCRATCH##*/}"
      - PGPASSWORD="$PROD_PASSWORD" pg_dump --schema-only --no-owner
        --no-privileges "$PROD" | psql -X -q -v ON_ERROR_STOP=1 -o /dev/null
        "$SCRATCH"
      - PGPASSWORD="$PROD_PASSWORD" pg_dump --data-only
        --table=databasechangelog "$PROD" | psql -X -q -v ON_ERROR_STOP=1 -o
        /dev/null "$SCRATCH"
      - psql -X -At "$SCRATCH" -c "SELECT count(*) || ' applied changesets
        copied to the scratch database' FROM databasechangelog"

  - id: rehearse
    type: io.kestra.plugin.liquibase.CLI
    description: "`update-testing-rollback` applies the pending changesets to the
      scratch copy, rolls them back, and applies them again. A changeset that
      fails, or whose rollback fails, stops the flow here, before anyone is
      asked to approve."
    containerImage: "{{ vars.liquibase_image }}"
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: host
      user: root
    inputFiles: "{{ outputs.plan.outputFiles | toJson }}"
    env:
      LIQUIBASE_COMMAND_URL: "jdbc:postgresql://{{ vars.scratch_db }}"
      LIQUIBASE_COMMAND_USERNAME: "{{ vars.scratch_user }}"
      LIQUIBASE_COMMAND_PASSWORD: "{{ secret('SCRATCH_DB_PASSWORD') }}"
      LIQUIBASE_COMMAND_CHANGELOG_FILE: "{{ vars.changelog_file }}"
      LIQUIBASE_SHOW_BANNER: "false"
      LIQUIBASE_ANALYTICS_ENABLED: "false"
    commands:
      - liquibase update-testing-rollback 2>&1

  - id: rehearsal_checks
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Runs the post-deployment checks on the rehearsed schema. The copy
      has no business data, so this only proves the checks still run against the
      new schema; their result is not used.
    url: "jdbc:postgresql://{{ vars.scratch_db }}"
    username: "{{ vars.scratch_user }}"
    password: "{{ secret('SCRATCH_DB_PASSWORD') }}"
    sql: "{{ read(outputs.plan.outputFiles[vars.verify_file]) }}"
    fetchType: FETCH

  - id: label_pending
    type: io.kestra.plugin.core.execution.Labels
    labels:
      outcome: awaiting-approval
      pending: "{{ outputs.plan.vars.pending }}"

  - id: show_plan
    type: io.kestra.plugin.core.log.Log
    message: |
      {{ outputs.plan.vars.pending }} changesets of {{ inputs.commit }} are not applied to {{ vars.prod_db }} yet. On a copy of its schema they applied, rolled back and applied again, and the post-deployment checks ran.
      SQL that APPLY runs on {{ vars.prod_db }} (the full script with Liquibase's own statements is preview.sql in the outputs of plan):
      {{ read(outputs.plan.outputFiles['changes.sql']) }}

  - id: approve
    type: io.kestra.plugin.core.flow.Pause
    description: Nothing touches production before someone resumes this execution.
      If nobody answers within `approval_window`, the execution is cancelled and
      production stays as it is.
    pauseDuration: "{{ vars.approval_window }}"
    behavior: CANCEL
    onResume:
      - id: decision
        type: SELECT
        displayName: Decision
        description: APPLY tags production, runs the changesets and checks the result,
          and rolls back to the tag if a check fails. REJECT changes nothing.
        isRadio: true
        expression: >-
          {{ [ {'label': 'APPLY: run ' ~ outputs.plan.vars.pending ~ '
          changesets on ' ~ vars.prod_db, 'value': 'APPLY'}, {'label': 'REJECT:
          leave ' ~ vars.prod_db ~ ' as it is', 'value': 'REJECT'} ] }}

      - id: note
        type: STRING
        displayName: Note
        description: Why you decided this (a change ticket, for example). Kept in the
          execution.
        required: false

  - id: rejected
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.approve.onResume.decision == 'REJECT' }}"
    then:
      - id: label_rejected
        type: io.kestra.plugin.core.execution.Labels
        labels:
          outcome: rejected

      - id: log_rejected
        type: io.kestra.plugin.core.log.Log
        level: WARN
        message: "Deploying {{ inputs.commit }} was rejected{{
          (outputs.approve.onResume.note ?? '') == '' ? '' : ': ' ~
          outputs.approve.onResume.note }}. Nothing was run on {{ vars.prod_db
          }}; {{ outputs.plan.vars.pending }} changesets stay pending."

      - id: stop_rejected
        type: io.kestra.plugin.core.execution.Exit
        state: WARNING

  - id: rollback_point
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: The rollback point is the latest applied changeset. Liquibase keeps
      one tag per changeset row and `tag` replaces it, so a tag already there (a
      release tag, say) is reused instead of overwritten.
    url: "jdbc:postgresql://{{ vars.prod_db }}"
    username: "{{ vars.prod_user }}"
    password: "{{ secret('PROD_DB_PASSWORD') }}"
    sql: SELECT coalesce((SELECT tag FROM databasechangelog ORDER BY orderexecuted
      DESC LIMIT 1), :fallback) AS tag
    parameters:
      fallback: "pre-{{ execution.id }}"
    fetchType: FETCH_ONE

  - id: tag_prod
    type: io.kestra.plugin.liquibase.CLI
    containerImage: "{{ vars.liquibase_image }}"
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: host
      user: root
    env:
      LIQUIBASE_COMMAND_URL: "jdbc:postgresql://{{ vars.prod_db }}"
      LIQUIBASE_COMMAND_USERNAME: "{{ vars.prod_user }}"
      LIQUIBASE_COMMAND_PASSWORD: "{{ secret('PROD_DB_PASSWORD') }}"
      LIQUIBASE_SHOW_BANNER: "false"
      LIQUIBASE_ANALYTICS_ENABLED: "false"
      TAG: "{{ outputs.rollback_point.row.tag }}"
    commands:
      - liquibase tag --tag="$TAG" 2>&1

  - id: update_prod
    type: io.kestra.plugin.liquibase.CLI
    containerImage: "{{ vars.liquibase_image }}"
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: host
      user: root
    inputFiles: "{{ outputs.plan.outputFiles | toJson }}"
    env:
      LIQUIBASE_COMMAND_URL: "jdbc:postgresql://{{ vars.prod_db }}"
      LIQUIBASE_COMMAND_USERNAME: "{{ vars.prod_user }}"
      LIQUIBASE_COMMAND_PASSWORD: "{{ secret('PROD_DB_PASSWORD') }}"
      LIQUIBASE_COMMAND_CHANGELOG_FILE: "{{ vars.changelog_file }}"
      LIQUIBASE_SHOW_BANNER: "false"
      LIQUIBASE_ANALYTICS_ENABLED: "false"
    commands:
      - liquibase update 2>&1

  - id: applied
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "jdbc:postgresql://{{ vars.prod_db }}"
    username: "{{ vars.prod_user }}"
    password: "{{ secret('PROD_DB_PASSWORD') }}"
    sql: |
      SELECT filename, id, author, exectype FROM databasechangelog
      WHERE orderexecuted > (SELECT max(orderexecuted) FROM databasechangelog WHERE tag = :tag)
      ORDER BY orderexecuted
    parameters:
      tag: "{{ outputs.rollback_point.row.tag }}"
    fetchType: FETCH

  - id: checks
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "jdbc:postgresql://{{ vars.prod_db }}"
    username: "{{ vars.prod_user }}"
    password: "{{ secret('PROD_DB_PASSWORD') }}"
    sql: "{{ read(outputs.plan.outputFiles[vars.verify_file]) }}"
    fetchType: FETCH

  - id: verify
    type: io.kestra.plugin.core.execution.Assert
    description: Every rehearsed changeset must be recorded as executed after the
      rollback point, and the post-deployment checks must return no rows.
      Otherwise the flow fails here and the `errors` branch rolls production
      back to the tag.
    conditions:
      - "{{ outputs.applied.size == outputs.plan.vars.pending and
        (outputs.applied.rows | jq('all(.exectype == \"EXECUTED\")') | first)
        }}"
      - "{{ outputs.checks.size == 0 }}"
    errorMessage: "Verification failed on {{ vars.prod_db }}: {{
      outputs.applied.size }} of {{ outputs.plan.vars.pending }} changesets
      executed, failed checks: {{ outputs.checks.rows | toJson }}"

  - id: label_applied
    type: io.kestra.plugin.core.execution.Labels
    labels:
      outcome: applied
      rollback_tag: "{{ outputs.rollback_point.row.tag }}"

  - id: log_summary
    type: io.kestra.plugin.core.log.Log
    message: |
      Applied {{ outputs.applied.size }} changesets of {{ inputs.commit }} to {{ vars.prod_db }}, approved{{ (outputs.approve.onResume.note ?? '') == '' ? '' : ' with the note "' ~ outputs.approve.onResume.note ~ '"' }}. The post-deployment checks found nothing. To undo it by hand: liquibase rollback --tag={{ outputs.rollback_point.row.tag }}
      {% for row in outputs.applied.rows %}{{ row.filename }}::{{ row.id }}::{{ row.author }}
      {% endfor %}

errors:
  - id: after_tag
    type: io.kestra.plugin.core.flow.If
    description: Production is rolled back only when the failure came after
      `tag_prod`. Before that, nothing was run on it.
    condition: "{{ (tasks.tag_prod.state ?? '') == 'SUCCESS' }}"
    then:
      - id: label_rollback_failed
        type: io.kestra.plugin.core.execution.Labels
        description: Set before the rollback, so the label stays if the rollback itself
          fails.
        labels:
          outcome: rollback-failed

      - id: rollback_prod
        type: io.kestra.plugin.liquibase.CLI
        containerImage: "{{ vars.liquibase_image }}"
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          networkMode: host
          user: root
        inputFiles: "{{ outputs.plan.outputFiles | toJson }}"
        env:
          LIQUIBASE_COMMAND_URL: "jdbc:postgresql://{{ vars.prod_db }}"
          LIQUIBASE_COMMAND_USERNAME: "{{ vars.prod_user }}"
          LIQUIBASE_COMMAND_PASSWORD: "{{ secret('PROD_DB_PASSWORD') }}"
          LIQUIBASE_COMMAND_CHANGELOG_FILE: "{{ vars.changelog_file }}"
          LIQUIBASE_SHOW_BANNER: "false"
          LIQUIBASE_ANALYTICS_ENABLED: "false"
          TAG: "{{ outputs.rollback_point.row.tag }}"
        commands:
          - liquibase rollback --tag="$TAG" 2>&1

      - id: after_rollback
        type: io.kestra.plugin.jdbc.postgresql.Query
        url: "jdbc:postgresql://{{ vars.prod_db }}"
        username: "{{ vars.prod_user }}"
        password: "{{ secret('PROD_DB_PASSWORD') }}"
        sql: SELECT filename, id FROM databasechangelog WHERE orderexecuted > (SELECT
          max(orderexecuted) FROM databasechangelog WHERE tag = :tag)
        parameters:
          tag: "{{ outputs.rollback_point.row.tag }}"
        fetchType: FETCH

      - id: rolled_back
        type: io.kestra.plugin.core.execution.Assert
        conditions:
          - "{{ outputs.after_rollback.size == 0 }}"
        errorMessage: "Rollback to {{ outputs.rollback_point.row.tag }} left {{
          outputs.after_rollback.size }} changesets applied on {{ vars.prod_db
          }}: {{ outputs.after_rollback.rows | toJson }}"

      - id: label_rolled_back
        type: io.kestra.plugin.core.execution.Labels
        labels:
          outcome: rolled-back
          rollback_tag: "{{ outputs.rollback_point.row.tag }}"

      - id: log_rolled_back
        type: io.kestra.plugin.core.log.Log
        level: ERROR
        message: "Deploying {{ inputs.commit }} failed in {{ (errorLogs() | last).taskId
          ?? 'an unknown task' }}: {{ (errorLogs() | last).message ?? 'see the
          execution logs' }}. {{ vars.prod_db }} was rolled back to {{
          outputs.rollback_point.row.tag }}, and no changeset of this deployment
          is left applied."
    else:
      - id: label_failed
        type: io.kestra.plugin.core.execution.Labels
        labels:
          outcome: failed

      - id: log_failed
        type: io.kestra.plugin.core.log.Log
        level: ERROR
        message: "Deploying {{ inputs.commit }} failed in {{ (errorLogs() | last).taskId
          ?? 'an unknown task' }}: {{ (errorLogs() | last).message ?? 'see the
          execution logs' }}. Nothing was run on {{ vars.prod_db }}."

triggers:
  - id: on_push
    type: io.kestra.plugin.core.trigger.Webhook
    description: Gitea posts every push. Only pushes to `branch` with a commit that
      touches `changelog_dir` start an execution; the others get a 204. The
      repository URL comes from `git_url`, never from the request.
    key: "{{ secret('MIGRATIONS_WEBHOOK_KEY') }}"
    when: >-
      {{ (trigger.body.ref ?? '') == 'refs/heads/' ~ vars.branch and
      (trigger.body | jq('[.commits[]? | (.added // []) + (.modified // []) +
      (.removed // []) | .[] | startswith("' ~ vars.changelog_dir ~ '/")] |
      any') | first) }}
    inputs:
      commit: "{{ trigger.body.after }}"
      pushed_by: "{{ trigger.body.pusher.login ?? '' }}"

A Liquibase changelog in Git is usually deployed by a CI job that runs liquibase update and hopes. This flow deploys it the careful way, on every push that touches the changelog: it rehearses the pending changesets on a scratch copy of the production schema, including their rollback, shows an approver the exact SQL, tags production, applies the changesets, runs your post-deployment checks against the real data, and rolls back to the tag if a check fails.

This blueprint was created by kruxarth.

How it works

on_push (Webhook) receives the Gitea push event. Only pushes to branch with a commit that touches changelog_dir start an execution. The commit SHA becomes the commit input; the repository URL always comes from git_url.
clone checks out that commit, and plan runs liquibase update-sql against production. The script, a copy without Liquibase's bookkeeping, and the changelog files are kept with the execution, so every later step uses exactly the files that were rehearsed.
Nothing pending: the execution ends with outcome: up-to-date.
rehearsal_copy recreates the scratch database from the production schema and the databasechangelog rows (no business data). rehearse runs liquibase update-testing-rollback there: apply, roll back, apply again. rehearsal_checks runs your check query on the result.
show_plan logs the SQL, and approve (Pause) waits for APPLY or REJECT and a note. REJECT ends in WARNING; no answer within approval_window cancels the execution.
rollback_point picks the tag, tag_prod runs liquibase tag, and update_prod runs liquibase update.
applied reads the changesets recorded after the tag, checks runs your check query on production, and verify (Assert) fails unless every rehearsed changeset was executed and no check returned a row.
Any failure after the tag runs liquibase rollback --tag in the errors branch, then checks that no changeset is left after the tag, and labels the execution outcome: rolled-back.

What you get

A changeset that fails, or whose rollback block fails, is caught on the scratch copy before anyone is asked to approve.
The approver sees the SQL that will run, and the decision and note stay in the execution.
Checks on real data after the apply. A changeset that runs fine but leaves wrong data, or fails halfway on production data, is rolled back to the tag automatically.
Labels outcome, commit, pushed_by, pending and rollback_tag on every execution.

Who it's for

Platform, DBA and backend teams that keep a Liquibase changelog in Git and deploy it to PostgreSQL.
Teams that want an approval and a record for production schema changes without building a release pipeline for them.

Why orchestrate this with Kestra

The Git server calls Kestra directly, and concurrency (limit 1) queues a second push until the first deployment is finished or rejected.
Pause with onResume inputs gives an approval form in the UI and the API, and the errors branch makes the rollback part of the flow, not a runbook.
Liquibase, pg_dump and the JDBC checks each run in their own container or plugin, and the whole process is one readable YAML file.

Prerequisites

Docker on the Kestra host with its socket available to Kestra. The task containers use the host network, so localhost is the Docker host.
A changelog repository with db/changelog.yaml and db/verify.sql: one SQL query whose every row is a failed check, for example SELECT 'orders with a negative total' AS failed_check, count(*) FROM orders WHERE total_cents < 0 HAVING count(*) > 0. Use UNION ALL to add checks.
The production PostgreSQL user that owns the schema (prod_user), and a scratch server where scratch_user has CREATEDB. The flow drops and recreates scratch_db on every run.
A Gitea (or Forgejo) push webhook, content type JSON, to http://<kestra>/api/v1/main/executions/webhook/company.platform/liquibase-migrate-on-push-with-rehearsal/<MIGRATIONS_WEBHOOK_KEY>. A Kestra on a private or loopback address needs [webhook] ALLOWED_HOST_LIST in Gitea's app.ini.

Secrets

MIGRATIONS_WEBHOOK_KEY: the key in the webhook URL. Anyone who knows it can start a deployment of git_url.
GIT_TOKEN: a read-only token of the git_username account, to clone the repository.
PROD_DB_PASSWORD: password of prod_user.
SCRATCH_DB_PASSWORD: password of scratch_user.

Quick start

Add the four secrets, set the variables (git_url, prod_db, scratch_db, users), and save the flow.
Add the webhook in the repository settings, then push a commit that adds a changeset.
The execution pauses. Read the SQL in the show_plan log, click Resume, choose APPLY and add a note.
When it succeeds, the labels show outcome: applied and the tag to roll back to by hand.

How to extend

Notify the approvers from onPause, or post the result back to the commit status.
Rehearse with data by dropping --schema-only, if your data may be copied.
Add --context-filter or --label-filter per environment and call the flow as a subflow for each one.

Deploy Liquibase changesets on a Git push, after a rehearsal, an approval and post-deployment checks with automatic rollback

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Deploy Liquibase changesets on a Git push, after a rehearsal, an approval and post-deployment checks with automatic rollback

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Kestra instance retention policy with dry run, approval and audit CSV

Release Kestra Flows from Git with Smoke Tests and Automatic Rollback

Kestra Orphaned Namespace Configuration Compliance Gate

Kestra Flow Security, Governance and Linting Gate

Purge aged Kestra execution history behind an approval gate

Re-enable Kestra triggers that were never switched back on

New to Kestra?