Reconcile PostgreSQL Grants Against a YAML Spec and Revoke Extras Only After Approval

id: postgres-grant-drift-reconciler
namespace: company.team
description: |
  Keep PostgreSQL privileges equal to a YAML spec: schema grants on tables and sequences, default
  privileges for objects created later, role memberships, and which roles may be SUPERUSER or
  BYPASSRLS. Missing grants are applied in one transaction. Extra grants (usually a manual GRANT)
  go to an approver as a multi-select; only the ones chosen are revoked, after the catalog is read
  again. Each change is reported once.

concurrency:
  limit: 1
  behavior: QUEUE

triggers:
  - id: after_migration
    type: io.kestra.plugin.core.trigger.Webhook
    key: grant-drift-change-me
    description: "Call this after a migration that creates tables, roles or grants:
      curl -X POST
      http://localhost:8080/api/v1/main/executions/webhook/company.team/postgre\
      s-grant-drift-reconciler/grant-drift-change-me"
  - id: hourly_drift_check
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "37 * * * *"
    disabled: true
    description: Shipped disabled. Enable it to catch changes made outside
      migrations, such as a manual GRANT.

inputs:
  - id: desired_grants
    type: STRING
    displayName: Desired grants (YAML)
    description: The spec the database must match. Edit it, or point it at a
      namespace file with read('grants.yaml'). Within the listed schemas, the
      listed roles get exactly these privileges, so anything else they hold
      there is an extra.
    defaults: |
      roles:
        - app_migrator
        - app_reader
        - app_writer
      privileged_roles:
        - postgres
      forbidden_roles:
        - legacy_admin
      schemas:
        - app
      grants:
        - role: app_reader
          schema: app
          tables: [SELECT]
          sequences: [SELECT]
        - role: app_writer
          schema: app
          tables: [SELECT, INSERT, UPDATE, DELETE]
          sequences: [USAGE]
      default_privileges:
        - role: app_reader
          owner: app_migrator
          schema: app
          tables: [SELECT]
          sequences: [SELECT]
        - role: app_writer
          owner: app_migrator
          schema: app
          tables: [SELECT, INSERT, UPDATE, DELETE]
          sequences: [USAGE]
      memberships:
        - member: app_writer
          role: app_reader

  - id: kestra_url
    type: STRING
    displayName: Kestra URL
    defaults: http://localhost:8080
    description: Base URL of your Kestra UI, used for the approval link in Slack.

variables:
  # One catalog read, used three times: before the approval, after it, and to store the state.
  # Parameters: :desired (the spec), :schemas, :privileged, :forbidden (JSON arrays), and
  # :chosen (the labels picked in the approval form). Every identifier is
  # quoted by PostgreSQL with format('%I'); privilege names come from the validated spec.
  catalog_sql: |
    WITH want AS (
      SELECT d.kind, d.grantee, d.owner, d.schema, d.priv, d.target
      FROM jsonb_to_recordset(CAST(:desired AS jsonb))
           AS d(kind TEXT, grantee TEXT, owner TEXT, schema TEXT, priv TEXT, target TEXT)
    ), managed AS (
      SELECT DISTINCT grantee FROM want WHERE kind IN ('table', 'sequence', 'default_table', 'default_sequence')
    ), managed_members AS (
      SELECT DISTINCT grantee FROM want WHERE kind = 'membership'
    ), managed_owners AS (
      SELECT DISTINCT owner FROM want WHERE kind IN ('default_table', 'default_sequence')
    ), rel AS (
      SELECT c.oid, c.relkind, CAST(n.nspname AS TEXT) AS schema, CAST(c.relname AS TEXT) AS relname,
             CAST(o.rolname AS TEXT) AS owner, c.relacl
      FROM pg_class c
      JOIN pg_namespace n ON n.oid = c.relnamespace
      JOIN pg_roles o ON o.oid = c.relowner
      WHERE c.relkind IN ('r', 'p', 'S')
        AND CAST(n.nspname AS TEXT) IN (SELECT jsonb_array_elements_text(CAST(:schemas AS jsonb)))
    ), acl AS (
      SELECT r.oid, r.schema, r.relkind, r.relname, r.owner,
             CAST(g.rolname AS TEXT) AS grantee, CAST(a.privilege_type AS TEXT) AS priv
      FROM rel r
      CROSS JOIN LATERAL aclexplode(r.relacl) a
      JOIN pg_roles g ON g.oid = a.grantee
    ), defacl AS (
      SELECT CAST(o.rolname AS TEXT) AS owner, CAST(n.nspname AS TEXT) AS schema,
             CASE WHEN d.defaclobjtype = 'S' THEN 'default_sequence' ELSE 'default_table' END AS kind,
             CAST(g.rolname AS TEXT) AS grantee, CAST(a.privilege_type AS TEXT) AS priv
      FROM pg_default_acl d
      JOIN pg_roles o ON o.oid = d.defaclrole
      JOIN pg_namespace n ON n.oid = d.defaclnamespace
      CROSS JOIN LATERAL aclexplode(d.defaclacl) a
      JOIN pg_roles g ON g.oid = a.grantee
      WHERE d.defaclobjtype IN ('r', 'S')
        AND CAST(n.nspname AS TEXT) IN (SELECT jsonb_array_elements_text(CAST(:schemas AS jsonb)))
    ), mem AS (
      SELECT CAST(m.rolname AS TEXT) AS grantee, CAST(t.rolname AS TEXT) AS target
      FROM pg_auth_members am
      JOIN pg_roles m ON m.oid = am.member
      JOIN pg_roles t ON t.oid = am.roleid
    ), found AS (
      SELECT 'MISSING_GRANT' AS finding, 'apply' AS action,
             format('%I %s on %s %I.%I', w.grantee, w.priv, CASE WHEN r.relkind = 'S' THEN 'sequence' ELSE 'table' END, r.schema, r.relname) AS label,
             format('GRANT %s ON %s %I.%I TO %I', w.priv, CASE WHEN r.relkind = 'S' THEN 'SEQUENCE' ELSE 'TABLE' END, r.schema, r.relname, w.grantee) AS statement
      FROM want w
      JOIN rel r ON r.schema = w.schema
      WHERE w.kind = CASE WHEN r.relkind = 'S' THEN 'sequence' ELSE 'table' END
        AND w.grantee <> r.owner
        AND NOT EXISTS (SELECT 1 FROM acl a WHERE a.oid = r.oid AND a.grantee = w.grantee AND a.priv = w.priv)
      UNION ALL
      SELECT 'MISSING_DEFAULT', 'apply',
             format('%I %s on new %s in schema %I created by %I', w.grantee, w.priv, CASE WHEN w.kind = 'default_sequence' THEN 'sequences' ELSE 'tables' END, w.schema, w.owner),
             format('ALTER DEFAULT PRIVILEGES FOR ROLE %I IN SCHEMA %I GRANT %s ON %s TO %I', w.owner, w.schema, w.priv, CASE WHEN w.kind = 'default_sequence' THEN 'SEQUENCES' ELSE 'TABLES' END, w.grantee)
      FROM want w
      WHERE w.kind IN ('default_table', 'default_sequence')
        AND NOT EXISTS (SELECT 1 FROM defacl da WHERE da.owner = w.owner AND da.schema = w.schema
                        AND da.kind = w.kind AND da.grantee = w.grantee AND da.priv = w.priv)
      UNION ALL
      SELECT 'MISSING_MEMBERSHIP', 'apply', format('%I is a member of %I', w.grantee, w.target),
             format('GRANT %I TO %I', w.target, w.grantee)
      FROM want w
      WHERE w.kind = 'membership'
        AND NOT EXISTS (SELECT 1 FROM mem m WHERE m.grantee = w.grantee AND m.target = w.target)
      UNION ALL
      SELECT 'EXTRA_GRANT', 'approve',
             format('%I %s on %s %I.%I', a.grantee, a.priv, CASE WHEN a.relkind = 'S' THEN 'sequence' ELSE 'table' END, a.schema, a.relname),
             format('REVOKE %s ON %s %I.%I FROM %I', a.priv, CASE WHEN a.relkind = 'S' THEN 'SEQUENCE' ELSE 'TABLE' END, a.schema, a.relname, a.grantee)
      FROM acl a
      WHERE a.grantee IN (SELECT grantee FROM managed)
        AND a.grantee <> a.owner
        AND NOT EXISTS (SELECT 1 FROM want w WHERE w.kind = CASE WHEN a.relkind = 'S' THEN 'sequence' ELSE 'table' END
                        AND w.schema = a.schema AND w.grantee = a.grantee AND w.priv = a.priv)
      UNION ALL
      SELECT 'EXTRA_DEFAULT', 'approve',
             format('%I %s on new %s in schema %I created by %I', da.grantee, da.priv, CASE WHEN da.kind = 'default_sequence' THEN 'sequences' ELSE 'tables' END, da.schema, da.owner),
             format('ALTER DEFAULT PRIVILEGES FOR ROLE %I IN SCHEMA %I REVOKE %s ON %s FROM %I', da.owner, da.schema, da.priv, CASE WHEN da.kind = 'default_sequence' THEN 'SEQUENCES' ELSE 'TABLES' END, da.grantee)
      FROM defacl da
      WHERE da.grantee IN (SELECT grantee FROM managed)
        AND da.owner IN (SELECT owner FROM managed_owners)
        AND da.grantee <> da.owner
        AND NOT EXISTS (SELECT 1 FROM want w WHERE w.kind = da.kind AND w.owner = da.owner AND w.schema = da.schema
                        AND w.grantee = da.grantee AND w.priv = da.priv)
      UNION ALL
      SELECT 'EXTRA_MEMBERSHIP', 'approve', format('%I is a member of %I', m.grantee, m.target),
             format('REVOKE %I FROM %I', m.target, m.grantee)
      FROM mem m
      WHERE m.grantee IN (SELECT grantee FROM managed_members)
        AND NOT EXISTS (SELECT 1 FROM want w WHERE w.kind = 'membership' AND w.grantee = m.grantee AND w.target = m.target)
      UNION ALL
      SELECT 'PRIVILEGED_ROLE', 'report',
             format('%I is %s', r.rolname, array_to_string(ARRAY[CASE WHEN r.rolsuper THEN 'SUPERUSER' END, CASE WHEN r.rolbypassrls THEN 'BYPASSRLS' END], ' and ')), NULL
      FROM pg_roles r
      WHERE (r.rolsuper OR r.rolbypassrls) AND left(r.rolname, 3) <> 'pg_'
        AND CAST(r.rolname AS TEXT) NOT IN (SELECT jsonb_array_elements_text(CAST(:privileged AS jsonb)))
      UNION ALL
      SELECT 'FORBIDDEN_ROLE', 'report', format('%I exists but the spec forbids it', r.rolname), NULL
      FROM pg_roles r
      WHERE CAST(r.rolname AS TEXT) IN (SELECT jsonb_array_elements_text(CAST(:forbidden AS jsonb)))
    )
    SELECT * FROM (
      SELECT f.finding, f.action, f.label, f.statement,
             (f.action = 'approve' AND f.label IN (SELECT jsonb_array_elements_text(CAST(:chosen AS jsonb)))) AS approved
      FROM found f
      UNION ALL
      SELECT 'ALREADY_GONE', 'gone', ch.label, NULL, true
      FROM jsonb_array_elements_text(CAST(:chosen AS jsonb)) AS ch(label)
      WHERE NOT EXISTS (SELECT 1 FROM found f WHERE f.action = 'approve' AND f.label = ch.label)
    ) x
    ORDER BY CASE x.action WHEN 'apply' THEN 1 WHEN 'report' THEN 2 WHEN 'approve' THEN 3 ELSE 4 END, x.label

  # Fingerprints of what was reported and what is extra, compared with the last stored state.
  # The state lives in the KV store of this namespace, so no table is needed in the database.
  state_sql: |
    WITH f AS (
      SELECT x.action, x.label
      FROM jsonb_to_recordset(CAST(:findings AS jsonb)) AS x(action TEXT, label TEXT)
    ), fp AS (
      SELECT 'report' AS scope, count(*) AS n, md5(coalesce(string_agg(label, E'\n' ORDER BY label), '')) AS fingerprint
      FROM f WHERE action = 'report'
      UNION ALL
      SELECT 'approve', count(*), md5(coalesce(string_agg(label, E'\n' ORDER BY label), ''))
      FROM f WHERE action = 'approve'
    )
    SELECT CAST(MAX(CASE WHEN scope = 'report' THEN fingerprint END) AS TEXT) AS report_fp,
           CAST(MAX(CASE WHEN scope = 'approve' THEN fingerprint END) AS TEXT) AS approve_fp,
           MAX(CASE WHEN scope = 'report' THEN n END) AS report_n,
           MAX(CASE WHEN scope = 'approve' THEN n END) AS approve_n,
           CAST(MAX(CASE WHEN scope = 'report' AND n > 0 AND fingerprint <> :report_known THEN 1 ELSE 0 END) AS INT) AS report_changed,
           CAST(MAX(CASE WHEN scope = 'approve' AND n > 0 AND fingerprint <> :approve_known THEN 1 ELSE 0 END) AS INT) AS approve_changed
    FROM fp

tasks:
  - id: parse_spec
    type: io.kestra.plugin.scripts.python.Script
    description: Parse the YAML and check its shape and privilege names. Python is
      used only for YAML; PyYAML is installed into a temporary directory on the
      worker.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    beforeCommands:
      - python3 -c "import yaml" 2>/dev/null || pip3 install -q
        --break-system-packages --target /tmp/kestra-pyyaml pyyaml
    env:
      PYTHONPATH: /tmp/kestra-pyyaml
    inputFiles:
      spec.yaml: "{{ inputs.desired_grants }}"
    script: |
      import json, sys
      import yaml

      TABLE_PRIVS = {'SELECT', 'INSERT', 'UPDATE', 'DELETE', 'TRUNCATE', 'REFERENCES', 'TRIGGER', 'MAINTAIN'}
      SEQUENCE_PRIVS = {'USAGE', 'SELECT', 'UPDATE'}
      KEYS = {'roles', 'privileged_roles', 'forbidden_roles', 'schemas', 'grants', 'default_privileges', 'memberships'}


      def emit(**outputs):
          print('::' + json.dumps({'outputs': outputs}) + '::')


      def fail(message):
          emit(valid=False, error=message.replace('::', ':').splitlines()[0][:300], desired='[]', roles='[]',
               schemas='[]', privileged='[]', forbidden='[]')
          sys.exit(0)


      try:
          with open('spec.yaml') as f:
              spec = yaml.safe_load(f)
      except yaml.YAMLError as e:
          fail('desired_grants is not valid YAML: ' + str(e))

      if not isinstance(spec, dict):
          fail('desired_grants must be a YAML mapping with roles, schemas, grants, default_privileges and memberships')
      unknown = sorted(set(spec) - KEYS)
      if unknown:
          fail('desired_grants has unknown key(s): ' + ', '.join(unknown))


      def names(key):
          values = spec.get(key) or []
          if not isinstance(values, list):
              fail(f'{key} must be a list of names')
          for v in values:
              if not isinstance(v, str) or not v or len(v.encode()) > 63 or '\n' in v or '\0' in v:
                  fail(f'{key} contains an invalid name: {v!r} (PostgreSQL names are 1-63 bytes, single line)')
          if len(set(values)) != len(values):
              fail(f'{key} lists the same name twice')
          return values


      roles = names('roles')
      schemas = names('schemas')
      privileged = names('privileged_roles')
      forbidden = names('forbidden_roles')


      def entries(key, required):
          values = spec.get(key) or []
          if not isinstance(values, list):
              fail(f'{key} must be a list')
          for item in values:
              if not isinstance(item, dict):
                  fail(f'every {key} entry must be a mapping')
              bad = sorted(set(item) - required)
              if bad:
                  fail(f'{key} entry has unknown key(s): ' + ', '.join(bad))
          return values


      def privs(item, key, allowed):
          values = item.get(key) or []
          if not isinstance(values, list):
              fail(f'{key} must be a list of privileges')
          out = []
          for p in values:
              p = str(p).upper()
              if p not in allowed:
                  fail(f'{key} privilege {p} is not allowed; use one of {", ".join(sorted(allowed))}')
              out.append(p)
          return out


      def known(name, what):
          if name not in roles:
              fail(f'{what} {name!r} is not listed under roles')


      def known_schema(name):
          if name not in schemas:
              fail(f'schema {name!r} is used in grants but not listed under schemas')


      desired = set()
      for g in entries('grants', {'role', 'schema', 'tables', 'sequences'}):
          role, schema = g.get('role'), g.get('schema')
          if not isinstance(role, str) or not isinstance(schema, str):
              fail('every grants entry needs a role and a schema')
          known(role, 'grants role')
          known_schema(schema)
          for p in privs(g, 'tables', TABLE_PRIVS):
              desired.add(('table', role, '', schema, p, ''))
          for p in privs(g, 'sequences', SEQUENCE_PRIVS):
              desired.add(('sequence', role, '', schema, p, ''))

      for d in entries('default_privileges', {'role', 'owner', 'schema', 'tables', 'sequences'}):
          role, owner, schema = d.get('role'), d.get('owner'), d.get('schema')
          if not all(isinstance(x, str) for x in (role, owner, schema)):
              fail('every default_privileges entry needs role, owner and schema')
          known(role, 'default_privileges role')
          known(owner, 'default_privileges owner')
          known_schema(schema)
          for p in privs(d, 'tables', TABLE_PRIVS):
              desired.add(('default_table', role, owner, schema, p, ''))
          for p in privs(d, 'sequences', SEQUENCE_PRIVS):
              desired.add(('default_sequence', role, owner, schema, p, ''))

      for m in entries('memberships', {'member', 'role'}):
          member, role = m.get('member'), m.get('role')
          if not isinstance(member, str) or not isinstance(role, str):
              fail('every memberships entry needs member and role')
          known(member, 'memberships member')
          known(role, 'memberships role')
          if member == role:
              fail(f'{member!r} cannot be a member of itself')
          desired.add(('membership', member, '', '', '', role))

      rows = [dict(zip(['kind', 'grantee', 'owner', 'schema', 'priv', 'target'], d)) for d in sorted(desired)]
      if not rows:
          fail('desired_grants defines no grants, default privileges or memberships')
      emit(valid=True, error='', desired=json.dumps(rows), roles=json.dumps(roles), schemas=json.dumps(schemas),
           privileged=json.dumps(privileged), forbidden=json.dumps(forbidden))
    retry:
      type: constant
      interval: PT2S
      maxAttempts: 2

  - id: spec_guard
    type: io.kestra.plugin.core.flow.If
    description: Stop before any database call when the spec is not valid.
    condition: "{{ outputs.parse_spec.vars.valid == false }}"
    then:
      - id: invalid_spec
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "{{ outputs.parse_spec.vars.error }}"

  - id: check_server
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Fail fast if the database is unreachable, and record which role
      runs the flow.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT current_database() AS database, CAST(current_user AS TEXT) AS as_role
    retry:
      type: exponential
      interval: PT5S
      maxInterval: PT1M
      maxAttempts: 3

  - id: check_objects
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Every role and schema in the spec must already exist. The flow
      never creates them.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    parameters:
      roles: "{{ outputs.parse_spec.vars.roles }}"
      schemas: "{{ outputs.parse_spec.vars.schemas }}"
    sql: |
      SELECT CAST(COALESCE((SELECT jsonb_agg(x) FROM jsonb_array_elements_text(CAST(:roles AS jsonb)) x
                            WHERE CAST(x AS TEXT) NOT IN (SELECT CAST(rolname AS TEXT) FROM pg_roles)), CAST('[]' AS jsonb)) AS TEXT) AS missing_roles,
             CAST(COALESCE((SELECT jsonb_agg(x) FROM jsonb_array_elements_text(CAST(:schemas AS jsonb)) x
                            WHERE CAST(x AS TEXT) NOT IN (SELECT CAST(nspname AS TEXT) FROM pg_namespace)), CAST('[]' AS jsonb)) AS TEXT) AS missing_schemas
    retry:
      type: exponential
      interval: PT5S
      maxInterval: PT1M
      maxAttempts: 3

  - id: objects_guard
    type: io.kestra.plugin.core.flow.If
    description: Missing roles or schemas stop the run before any change.
    condition: "{{ (fromJson(outputs.check_objects.row.missing_roles) | length) +
      (fromJson(outputs.check_objects.row.missing_schemas) | length) > 0 }}"
    then:
      - id: missing_objects
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "The spec names objects that do not exist in {{
          outputs.check_server.row.database }}: roles {{
          outputs.check_objects.row.missing_roles }}, schemas {{
          outputs.check_objects.row.missing_schemas }}. Create them, or fix the
          spec."

  - id: diff
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Read the catalog and compare it with the spec. Each row is one
      finding with action apply (missing, fixed automatically), approve (extra,
      needs approval) or report (only shown).
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    parameters:
      desired: "{{ outputs.parse_spec.vars.desired }}"
      schemas: "{{ outputs.parse_spec.vars.schemas }}"
      privileged: "{{ outputs.parse_spec.vars.privileged }}"
      forbidden: "{{ outputs.parse_spec.vars.forbidden }}"
      chosen: "[]"
    sql: "{{ vars.catalog_sql }}"
    retry:
      type: exponential
      interval: PT5S
      maxInterval: PT1M
      maxAttempts: 3

  - id: state_check
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Decide what is new since the last run. A set of extras or report
      lines that was already reported is not reported again.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    parameters:
      findings: "{{ outputs.diff.rows | toJson }}"
      report_known: "{{ kv('grant-drift-report-state', errorOnMissing=false) ?? '' }}"
      approve_known: "{{ kv('grant-drift-approve-state', errorOnMissing=false) ?? '' }}"
    sql: "{{ vars.state_sql }}"

  - id: apply_gate
    type: io.kestra.plugin.core.flow.If
    description: Only run the transaction when something is missing.
    condition: "{{ outputs.diff.rows | jq('[.[] | select(.action == \"apply\")] |
      length') | first > 0 }}"
    then:
      - id: apply_missing
        type: io.kestra.plugin.jdbc.postgresql.Queries
        description: All missing GRANT, ALTER DEFAULT PRIVILEGES and GRANT role
          statements run in one transaction. If one fails, none of them is kept.
        url: "{{ secret('POSTGRES_URL') }}"
        username: "{{ secret('POSTGRES_USER') }}"
        password: "{{ secret('POSTGRES_PASSWORD') }}"
        transaction: true
        sql: "{{ outputs.diff.rows | jq('[.[] | select(.action == \"apply\") |
          .statement] | join(\"; \")') | first }}"
        retry:
          type: exponential
          interval: PT5S
          maxInterval: PT1M
          maxAttempts: 2
      - id: notify_applied
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: One message for the change that was made.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          PostgreSQL {{ outputs.check_server.row.database }}: applied {{ outputs.diff.rows | jq('[.[] | select(.action == "apply")] | length') | first }} missing grant(s) in one transaction, as the spec requires:
          {% for r in outputs.diff.rows %}{% if r.action == 'apply' %}- {{ r.label }}
          {% endif %}{% endfor %}Run {{ inputs.kestra_url }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }} for details.

  - id: report_gate
    type: io.kestra.plugin.core.flow.If
    description: Report privileged or forbidden roles once, when the list changes.
      Nothing is changed for these.
    condition: "{{ outputs.state_check.row.report_changed == 1 }}"
    then:
      - id: notify_report
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          PostgreSQL {{ outputs.check_server.row.database }}: roles that need a human decision (nothing was changed):
          {% for r in outputs.diff.rows %}{% if r.action == 'report' %}- {{ r.label }}
          {% endif %}{% endfor %}To keep a superuser, add the role to privileged_roles in the spec. Otherwise run ALTER ROLE ... NOSUPERUSER NOBYPASSRLS, or drop the forbidden role by hand.

  - id: approval_gate
    type: io.kestra.plugin.core.flow.If
    description: Extra privileges are never revoked without a decision. A set that
      was already offered is not offered again until it changes.
    condition: "{{ outputs.state_check.row.approve_changed == 1 }}"
    then:
      - id: ask_on_call
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Lists the extras. The approval form offers the same labels. Choose
          within 2 hours, or nothing is revoked.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          PostgreSQL {{ outputs.check_server.row.database }}: {{ outputs.state_check.row.approve_n }} grant(s) not in the spec. Choose within 2 hours, or nothing is revoked: {{ inputs.kestra_url }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
          {% for r in outputs.diff.rows %}{% if r.action == 'approve' %}- {{ r.label }}
          {% endif %}{% endfor %}
      - id: wait_for_decision
        type: io.kestra.plugin.core.flow.Pause
        description: No answer within 2 hours means nothing is revoked (behavior RESUME
          with no selection).
        pauseDuration: PT2H
        behavior: RESUME
        onResume:
          - id: actions
            type: MULTISELECT
            displayName: Extras to revoke
            expression: "{{ outputs.diff.rows | jq('[.[] | select(.action == \"approve\") |
              .label]') | first }}"
            required: false

      - id: reread
        type: io.kestra.plugin.jdbc.postgresql.Query
        description: Read the catalog again after the approval. Only chosen extras that
          still exist are revoked.
        url: "{{ secret('POSTGRES_URL') }}"
        username: "{{ secret('POSTGRES_USER') }}"
        password: "{{ secret('POSTGRES_PASSWORD') }}"
        fetchType: FETCH
        parameters:
          desired: "{{ outputs.parse_spec.vars.desired }}"
          schemas: "{{ outputs.parse_spec.vars.schemas }}"
          privileged: "{{ outputs.parse_spec.vars.privileged }}"
          forbidden: "{{ outputs.parse_spec.vars.forbidden }}"
          chosen: "{{ (outputs.wait_for_decision.onResume.actions ?? []) | toJson }}"
        sql: "{{ vars.catalog_sql }}"

      - id: revoke_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.reread.rows | jq('[.[] | select(.action == \"approve\"
          and .approved == true)] | length') | first > 0 }}"
        then:
          - id: revoke_extras
            type: io.kestra.plugin.core.flow.Loop
            description: One REVOKE per chosen extra. Each runs alone, so one failure does
              not stop the others.
            values: "{{ outputs.reread.rows | jq('[.[] | select(.action == \"approve\" and
              .approved == true)]') | first | toJson }}"
            tasks:
              - id: revoke_one
                type: io.kestra.plugin.core.flow.Sequential
                allowFailure: true
                tasks:
                  - id: revoke
                    type: io.kestra.plugin.jdbc.postgresql.Query
                    url: "{{ secret('POSTGRES_URL') }}"
                    username: "{{ secret('POSTGRES_USER') }}"
                    password: "{{ secret('POSTGRES_PASSWORD') }}"
                    sql: "{{ fromJson(item.value).statement }}"

  - id: recheck
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Read the catalog once more at the end. This is the state that is
      stored and the one the outcome message reports.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    parameters:
      desired: "{{ outputs.parse_spec.vars.desired }}"
      schemas: "{{ outputs.parse_spec.vars.schemas }}"
      privileged: "{{ outputs.parse_spec.vars.privileged }}"
      forbidden: "{{ outputs.parse_spec.vars.forbidden }}"
      chosen: "{% if outputs.wait_for_decision is defined %}{{
        (outputs.wait_for_decision.onResume.actions ?? []) | toJson }}{% else
        %}[]{% endif %}"
    sql: "{{ vars.catalog_sql }}"

  - id: state_after
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Fingerprints of what is left after the run, to store.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    parameters:
      findings: "{{ outputs.recheck.rows | toJson }}"
      report_known: ""
      approve_known: ""
    sql: "{{ vars.state_sql }}"

  - id: remember_report
    type: io.kestra.plugin.core.kv.Set
    description: Store the report fingerprint, so the same report is not sent again.
    key: grant-drift-report-state
    kvType: STRING
    value: "{{ outputs.state_after.row.report_fp }}"

  - id: remember_approve
    type: io.kestra.plugin.core.kv.Set
    description: Store the extras fingerprint after the run. A declined set stays
      stored and is not offered again.
    key: grant-drift-approve-state
    kvType: STRING
    value: "{{ outputs.state_after.row.approve_fp }}"

  - id: outcome_gate
    type: io.kestra.plugin.core.flow.If
    description: Report the outcome only when something was chosen.
    condition: "{{ outputs.reread is defined and (outputs.reread.rows | jq('[.[] |
      select((.action == \"approve\" and .approved == true) or .action ==
      \"gone\")] | length') | first) > 0 }}"
    then:
      - id: notify_outcome
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          PostgreSQL {{ outputs.check_server.row.database }}: revoke result.
          Revoked: {{ {'r': outputs.reread.rows, 'k': outputs.recheck.rows} | jq('([.r[] | select(.action == "approve" and .approved == true) | .label]) - [.k[] | select(.action == "approve") | .label] | join(", ")') | first }}
          Already gone before the revoke: {{ outputs.reread.rows | jq('[.[] | select(.action == "gone") | .label] | join(", ")') | first }}
          Still present, not revoked (see the execution logs): {{ outputs.recheck.rows | jq('[.[] | select(.action == "approve" and .approved == true) | .label] | join(", ")') | first }}

  - id: partial_gate
    type: io.kestra.plugin.core.flow.If
    description: An approved extra that is still present after the run ends the run
      WARNING, so a partial failure is not reported as success.
    condition: "{{ outputs.recheck.rows | jq('[.[] | select(.action == \"approve\"
      and .approved == true)] | length') | first > 0 }}"
    then:
      - id: still_present
        type: io.kestra.plugin.core.execution.Exit
        state: WARNING

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: >-
      PostgreSQL grant drift reconciler failed in execution {{ execution.id }}
      {% if errorLogs() | length > 0 %}at `{{ errorLogs()[0]['taskId'] }}`: {{
      errorLogs()[0]['message'] | split(' \[\[') | first }}{% else %}without an
      error message{% endif %}. Grants applied before the failure are listed in
      the execution: {{ inputs.kestra_url }}/ui/main/executions/{{
      flow.namespace }}/{{ flow.id }}/{{ execution.id }}

outputs:
  - id: findings
    type: JSON
    description: The final catalog findings, one row per missing, extra or reported item.
    value: "{{ outputs.recheck.rows ?? [] }}"