PostgreSQL Unvalidated Constraint Validation Guard

id: postgres-unvalidated-constraint-validation-guard
namespace: company.team

inputs:
  - id: postgres_url
    type: STRING
    defaults: jdbc:postgresql://localhost:5432/postgres
    description: >-
      PostgreSQL JDBC URL for the database whose constraints are checked.
      Example jdbc:postgresql://postgres:5432/app.

  - id: use_secret_auth
    type: BOOL
    defaults: true
    description: >-
      Leave true in production so the connection reads POSTGRES_USERNAME and
      POSTGRES_PASSWORD from secrets. Set false only to pass credentials inline
      in a disposable test environment.

  - id: postgres_username
    type: STRING
    defaults: postgres
    description: Username used only when use_secret_auth is false. Ignored in production.

  - id: postgres_password
    type: STRING
    defaults: ""
    description: Password used only when use_secret_auth is false. Ignored in production.

  - id: schema
    type: STRING
    defaults: ""
    description: >-
      Optional single schema to restrict the scan to. Leave empty to scan every
      user schema in the database.

  - id: constraint_type
    type: SELECT
    values:
      - all
      - foreign_key
      - check
    defaults: all
    description: >-
      Which constraint kinds to look at. all covers FOREIGN KEY plus CHECK, the
      two kinds PostgreSQL lets you add NOT VALID.

  - id: max_table_mb
    type: INT
    defaults: 0
    description: >-
      Size ceiling for a table whose constraint may be validated in this run, in
      MB of table data. Validation scans the whole table, so a ceiling keeps a
      long scan out of an unattended window. 0 means no ceiling. Anything above
      the ceiling is still reported, it is only left out of the validation set.

  - id: dry_run
    type: BOOL
    defaults: true
    description: When true, report the unvalidated constraints without validating anything.

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: >-
      When true and dry_run is false, run ALTER TABLE VALIDATE CONSTRAINT behind
      an approval gate, then re-read pg_constraint.

variables:
  detect_sql: |
    WITH unvalidated AS (
      SELECT n.nspname AS schema_name,
             t.relname AS table_name,
             c.conname AS constraint_name,
             CASE c.contype WHEN 'f' THEN 'FOREIGN KEY' WHEN 'c' THEN 'CHECK' ELSE 'OTHER' END AS kind,
             pg_get_constraintdef(c.oid) AS condef,
             pg_table_size(c.conrelid) AS table_bytes,
             pg_size_pretty(pg_table_size(c.conrelid)) AS table_size,
             GREATEST(t.reltuples, 0)::bigint AS est_rows
      FROM pg_constraint c
      JOIN pg_class t ON t.oid = c.conrelid
      JOIN pg_namespace n ON n.oid = t.relnamespace
      WHERE NOT c.convalidated
        AND c.contype IN ('f', 'c')
        AND t.relkind IN ('r', 'p')
        AND ('{{ inputs.constraint_type }}' = 'all'
             OR ('{{ inputs.constraint_type }}' = 'foreign_key' AND c.contype = 'f')
             OR ('{{ inputs.constraint_type }}' = 'check' AND c.contype = 'c'))
        AND n.nspname NOT IN ('pg_catalog', 'information_schema')
        AND n.nspname NOT LIKE 'pg\_%'
        AND ('{{ inputs.schema }}' = '' OR n.nspname = '{{ inputs.schema }}')
    )
    SELECT schema_name, table_name, constraint_name, kind, condef,
      table_bytes, table_size, est_rows,
      (schema_name || '.' || table_name || '.' || constraint_name) AS qname,
      CASE WHEN {{ inputs.max_table_mb }} <= 0
                OR table_bytes <= ({{ inputs.max_table_mb }}::bigint * 1024 * 1024)
           THEN 'yes' ELSE 'no' END AS in_scope,
      format('ALTER TABLE %I.%I VALIDATE CONSTRAINT %I', schema_name, table_name, constraint_name) AS validate_sql,
      (schema_name || '.' || table_name || ' ' || constraint_name || ' (' || kind || ' on ' || table_size || ' table, about ' || est_rows || ' rows)') AS label
    FROM unvalidated
    ORDER BY table_bytes DESC, schema_name, table_name, constraint_name

triggers:
  - id: daily_unvalidated_constraint_check
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "0 4 * * *"
    description: >-
      Daily sweep for constraints left NOT VALID. Disabled by default and pinned
      to safe mode so it reports without validating anything until an operator
      enables it.
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: detect
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: >-
      Read pg_constraint for FOREIGN KEY and CHECK constraints whose
      convalidated is false. Each row carries the constraint definition, the
      table data size, the estimated row count, the exact ALTER TABLE VALIDATE
      CONSTRAINT statement and whether the table is inside the max_table_mb
      ceiling.
    url: "{{ inputs.postgres_url }}"
    username: "{{ inputs.use_secret_auth ? secret('POSTGRES_USERNAME') :
      inputs.postgres_username }}"
    password: "{{ inputs.use_secret_auth ? secret('POSTGRES_PASSWORD') :
      inputs.postgres_password }}"
    fetchType: FETCH
    sql: "{{ render(vars.detect_sql) }}"

  - id: summarize
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      Build the human readable lists once, so the log, the alert, the approval
      gate and the validation loop all read the same set.
    values:
      names: >-
        {{ outputs.detect.rows | jq('map(.label) | join(" | ")') | first }}
      target_names: >-
        {{ outputs.detect.rows | jq('map(select(.in_scope == "yes")) |
        map(.label) | join(" | ")') | first }}
      skipped_names: >-
        {{ outputs.detect.rows | jq('map(select(.in_scope == "no")) |
        map(.label) | join(" | ")') | first }}
      targets_json: >-
        {{ outputs.detect.rows | jq('map(select(.in_scope == "yes"))') | first |
        toJson }}

  - id: log_detect
    type: io.kestra.plugin.core.log.Log
    message: |
      PostgreSQL constraints still NOT VALID: {{ outputs.detect.size }} found.
      Found: {{ outputs.summarize.values.names }}
      Inside the {{ inputs.max_table_mb }} MB ceiling: {{ outputs.summarize.values.target_names }}
      Above the ceiling, reported only: {{ outputs.summarize.values.skipped_names }}
      Mode: dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }} constraint_type={{ inputs.constraint_type }} max_table_mb={{ inputs.max_table_mb }}

  - id: guard
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.detect.size > 0 }}"
    then:
      - id: findings_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: >-
          Alert on every detection, in dry run and in remediate mode, so a human
          sees the unverified rules even when nothing is validated.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"text": "PostgreSQL unvalidated constraint guard: {{ outputs.detect.size }} constraint(s) are enforced for new rows but were never checked against existing rows. Existing data may already break them and the planner will not use an unvalidated CHECK for constraint exclusion. Found {{ outputs.summarize.values.names }}. dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}."}

      - id: mode
        type: io.kestra.plugin.core.flow.If
        condition: >-
          {{ inputs.auto_remediate and not inputs.dry_run and
          (outputs.detect.rows | jq('map(select(.in_scope == "yes")) | length')
          | first) > 0 }}
        then:
          - id: approval_gate
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              Hold the run until a human approves the validation. ALTER TABLE
              VALIDATE CONSTRAINT takes a SHARE UPDATE EXCLUSIVE lock on the
              table and reads every row, so it competes with DDL and with
              autovacuum for the length of a full table scan. The table size and
              the estimated row count of each target are in the log above, which
              is the cost of approving.
            pauseDuration: PT2H
            behavior: FAIL
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: >-
                  Set true to validate the listed constraints. Anything else
                  leaves them NOT VALID.

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval_gate.onResume.approved ?? false }}"
            then:
              - id: validate_constraints
                type: io.kestra.plugin.core.flow.Loop
                description: >-
                  Remediation loop. One ALTER TABLE VALIDATE CONSTRAINT per
                  target, concurrencyLimit 1 so the table scans run in turn
                  rather than competing for the same table lock. allowFailure so
                  a constraint that existing rows genuinely break does not stop
                  the pass. The re-read of pg_constraint stays the real gate.
                concurrencyLimit: 1
                values: "{{ outputs.summarize.values.targets_json }}"
                tasks:
                  - id: validate_one
                    type: io.kestra.plugin.jdbc.postgresql.Query
                    description: >-
                      Validate one constraint against the rows already in the
                      table. PostgreSQL rejects this with a check or foreign key
                      violation when the existing data breaks the rule, which is
                      a finding about the data rather than a flow error.
                    allowFailure: true
                    url: "{{ inputs.postgres_url }}"
                    username: "{{ inputs.use_secret_auth ? secret('POSTGRES_USERNAME') :
                      inputs.postgres_username }}"
                    password: "{{ inputs.use_secret_auth ? secret('POSTGRES_PASSWORD') :
                      inputs.postgres_password }}"
                    fetchType: NONE
                    sql: "{{ fromJson(item.value).validate_sql }}"

                  - id: validate_log
                    type: io.kestra.plugin.core.log.Log
                    message: "Ran {{ fromJson(item.value).validate_sql }} for {{
                      fromJson(item.value).condef }}"

              - id: reverify
                type: io.kestra.plugin.jdbc.postgresql.Query
                description: >-
                  Re-read pg_constraint with the same query. Any constraint that
                  was a target and still reports convalidated false is a
                  failure.
                url: "{{ inputs.postgres_url }}"
                username: "{{ inputs.use_secret_auth ? secret('POSTGRES_USERNAME') :
                  inputs.postgres_username }}"
                password: "{{ inputs.use_secret_auth ? secret('POSTGRES_PASSWORD') :
                  inputs.postgres_password }}"
                fetchType: FETCH
                sql: "{{ render(vars.detect_sql) }}"

              - id: verify_plan
                type: io.kestra.plugin.core.output.OutputValues
                description: >-
                  Keep only the targets that are still NOT VALID after the pass.
                  A constraint added NOT VALID during the run is excluded by
                  design. So is anything left out by the size ceiling.
                values:
                  still_json: >-
                    {{ {'t': outputs.detect.rows, 'r': outputs.reverify.rows} |
                    toJson | jq('(.r | map(.qname)) as $rem | (.t |
                    map(select(.in_scope == "yes"))) | map(select(([.qname] -
                    $rem) | length == 0))') | first | toJson }}
                  still_names: >-
                    {{ {'t': outputs.detect.rows, 'r': outputs.reverify.rows} |
                    toJson | jq('(.r | map(.qname)) as $rem | (.t |
                    map(select(.in_scope == "yes"))) | map(select(([.qname] -
                    $rem) | length == 0) | .qname + " " + .condef) | join(" |
                    ")') | first }}

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (outputs.verify_plan.values.still_json | jq('length') | first) >
                  0 }}"
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      {{ outputs.verify_plan.values.still_json | jq('length') |
                      first }} targeted constraint(s) are still NOT VALID after
                      the pass: {{ outputs.verify_plan.values.still_names }}.
                      PostgreSQL refuses to validate a constraint that rows
                      already in the table break, so read this as a data finding
                      and fix or quarantine those rows. The per-constraint
                      PostgreSQL error is in the loop sub-execution log. A
                      missing table owner privilege gives the same outcome.
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: "Validated and verified. {{ outputs.detect.size }} constraint(s) were
                      NOT VALID before, 0 of the targeted ones after.
                      convalidated is now true for {{
                      outputs.summarize.values.target_names }}."
            else:
              - id: not_approved
                type: io.kestra.plugin.core.log.Log
                message: "Approval not granted, so nothing was validated. {{ outputs.detect.size
                  }} constraint(s) are still NOT VALID."
        else:
          - id: report_only
            type: io.kestra.plugin.core.log.Log
            message: "Report only (dry_run={{ inputs.dry_run }}, auto_remediate={{
              inputs.auto_remediate }}, max_table_mb={{ inputs.max_table_mb }}).
              {{ outputs.detect.size }} constraint(s) are NOT VALID: {{
              outputs.summarize.values.names }}. Of those, these would be
              validated: {{ outputs.summarize.values.target_names }}. No change
              made."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "Every FOREIGN KEY and CHECK constraint in scope reports convalidated
          true, so each one has been checked against the rows already in its
          table. Nothing to validate."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: >-
      Post to Slack on any failure, including a connection error or targeted
      constraints still NOT VALID after the pass.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "PostgreSQL unvalidated constraint guard FAILED (execution {{ execution.id }}). Either the database was unreachable or a targeted constraint is still NOT VALID after the pass, which usually means existing rows break the rule. Check the run."}