Automated PostgreSQL Backup to S3 with Integrity Verification Gate

id: postgres-backup-s3-retention-gate
namespace: company.team
description: |
  Automate PostgreSQL database logical backups with pg_dump and gzip, stream
  compressed archives directly to Amazon S3 with timestamped pathing, verify
  archive size and table count integrity, persist operational metadata to KV,
  and alert Slack immediately on anomalies or pipeline failures.

triggers:
  - id: daily_backup_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Runs daily database backup at 02:00 UTC. Shipped disabled; enable
      after your first successful manual verification.
    cron: "0 2 * * *"
    disabled: true

  - id: on_demand_webhook_backup
    type: io.kestra.plugin.core.trigger.Webhook
    description: Receives authenticated webhook calls for automated pre-deployment
      database snapshots.
    key: "{{ secret('BACKUP_WEBHOOK_KEY') }}"

inputs:
  - id: database_name
    type: STRING
    defaults: app_production
    description: Target PostgreSQL database name to back up.

  - id: min_backup_size_kb
    type: INT
    defaults: 10
    description: Minimum expected backup archive size in KB. Smaller archives trip
      the integrity alert.

  - id: s3_bucket
    type: STRING
    defaults: company-postgres-backups
    description: Amazon S3 bucket name designated for database archives.

  - id: aws_region
    type: STRING
    defaults: us-east-1
    description: AWS region hosting the destination S3 bucket.

tasks:
  - id: check_db_connectivity
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Verify PostgreSQL connectivity and fetch table count from the
      public schema prior to starting the dump.
    url: "jdbc:postgresql://{{ secret('POSTGRES_HOST') }}:5432/{{
      inputs.database_name }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      SELECT count(*) as table_count FROM information_schema.tables WHERE table_schema = 'public';
    fetchType: FETCH_ONE

  - id: generate_backup
    type: io.kestra.plugin.scripts.shell.Commands
    description: Execute pg_dump inside a postgres-client container, compress output
      with gzip, and calculate archive size.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      image: postgres:17-alpine
    env:
      PGHOST: "{{ secret('POSTGRES_HOST') }}"
      PGPORT: "5432"
      PGDATABASE: "{{ inputs.database_name }}"
      PGUSER: "{{ secret('POSTGRES_USER') }}"
      PGPASSWORD: "{{ secret('POSTGRES_PASSWORD') }}"
    outputFiles:
      - backup.sql.gz
    commands:
      - pg_dump -h "$PGHOST" -p "$PGPORT" -U "$PGUSER" "$PGDATABASE" | gzip >
        backup.sql.gz
      - BACKUP_SIZE_KB=$(du -k backup.sql.gz | cut -f1)
      - echo "Generated backup size: ${BACKUP_SIZE_KB} KB"
      - echo "::{\"outputs\":{\"backup_size_kb\": ${BACKUP_SIZE_KB}}}::"

  - id: upload_to_s3
    type: io.kestra.plugin.aws.s3.Upload
    description: Stream the compressed database backup archive directly to Amazon S3
      with ISO date partitioning.
    from: "{{ outputs.generate_backup.outputFiles['backup.sql.gz'] }}"
    bucket: "{{ inputs.s3_bucket }}"
    region: "{{ inputs.aws_region }}"
    key: "backups/{{ inputs.database_name }}/{{ now() | date('yyyy-MM-dd')
      }}/backup-{{ execution.startDate | date('yyyyMMdd-HHmmss') }}.sql.gz"
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"

  - id: verify_backup_integrity
    type: io.kestra.plugin.core.flow.If
    description: Evaluate archive size against the minimum threshold to catch empty
      or partial dumps before acknowledging success.
    condition: "{{ outputs.generate_backup.vars.backup_size_kb >=
      inputs.min_backup_size_kb }}"
    then:
      - id: record_latest_backup_kv
        type: io.kestra.plugin.core.kv.Set
        description: Persist latest backup metadata into namespace KV for operational
          auditing and automated restore pipelines.
        key: POSTGRES_LAST_BACKUP_METADATA
        value: |
          {
            "database": "{{ inputs.database_name }}",
            "s3_key": "{{ outputs.upload_to_s3.key }}",
            "size_kb": {{ outputs.generate_backup.vars.backup_size_kb }},
            "table_count": {{ outputs.check_db_connectivity.row.table_count }},
            "completed_at": "{{ execution.startDate }}"
          }

      - id: log_backup_success
        type: io.kestra.plugin.core.log.Log
        description: Record structured operational log for audit tracking.
        message: "PostgreSQL backup SUCCESS: database '{{ inputs.database_name }}' ({{
          outputs.generate_backup.vars.backup_size_kb }} KB, {{
          outputs.check_db_connectivity.row.table_count }} tables) archived to
          s3://{{ inputs.s3_bucket }}/{{ outputs.upload_to_s3.key }}."
    else:
      - id: alert_underweight_backup
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alert Slack when backup archive size falls below threshold,
          indicating probable data omission.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":warning: *PostgreSQL Backup Anomaly*: Backup for database `{{ inputs.database_name }}` produced only {{ outputs.generate_backup.vars.backup_size_kb }} KB (threshold: {{ inputs.min_backup_size_kb }} KB). Execution {{ execution.id }}."
          }

      - id: fail_on_anomaly
        type: io.kestra.plugin.core.execution.Fail
        description: Fail execution to reflect abnormal backup size.
        errorMessage: "Backup size {{ outputs.generate_backup.vars.backup_size_kb }} KB
          is below minimum threshold {{ inputs.min_backup_size_kb }} KB."

errors:
  - id: alert_pipeline_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack immediately if any task in the backup pipeline
      encounters an unexpected failure.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": ":rotating_light: *PostgreSQL Backup Pipeline FAILED* in flow `{{ flow.id }}` (execution `{{ execution.id }}`). Database `{{ inputs.database_name }}` backup failed — inspect Kestra execution logs."
      }

outputs:
  - id: backup_summary
    type: JSON
    description: 'Comprehensive backup metadata including S3 key, size in KB, public
      table count, and database name.'
    value: |
      {
        "database": "{{ inputs.database_name }}",
        "s3_key": "{{ outputs.upload_to_s3.key }}",
        "size_kb": {{ outputs.generate_backup.vars.backup_size_kb }},
        "table_count": {{ outputs.check_db_connectivity.row.table_count }},
        "status": "COMPLETED"
      }