Refresh PostgreSQL materialized views in dependency order

id: postgres-matview-dependency-refresh
namespace: company.data

inputs:
  - id: views
    type: JSON
    defaults:
      - schema: analytics
        name: daily_sales
      - schema: analytics
        name: sales_summary
    description: Explicit list of existing materialized views, including every
      upstream materialized view. Maximum 25.
  - id: dry_run
    type: BOOL
    defaults: true
    description: Inspect the ordered plan without refreshing any view.

concurrency:
  limit: 1
  behavior: QUEUE

tasks:
  - id: plan
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Follow catalog dependencies through ordinary views; reject an
      incomplete or unsafe refresh set.
    url: "{{ secret('POSTGRES_JDBC_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    parameters:
      requested: "{{ inputs.views | toJson }}"
    sql: |
      WITH RECURSIVE
      document AS (SELECT CAST(:requested AS jsonb) AS payload),
      requested AS (
        SELECT entry, entry->>'schema' AS schema_name, entry->>'name' AS view_name
        FROM document, jsonb_array_elements(payload) AS entry
      ),
      selected AS (
        SELECT r.*, c.oid, c.relkind, c.relispopulated,
               format('%I.%I', r.schema_name, r.view_name) AS qualified_name
        FROM requested r
        LEFT JOIN pg_namespace n ON n.nspname = r.schema_name
        LEFT JOIN pg_class c ON c.relnamespace = n.oid AND c.relname = r.view_name
      ),
      edges AS (
        SELECT DISTINCT rw.ev_class AS dependent, d.refobjid AS upstream
        FROM pg_rewrite rw
        JOIN pg_depend d ON d.classid = CAST('pg_rewrite' AS regclass) AND d.objid = rw.oid
        JOIN pg_class c ON c.oid = d.refobjid AND c.relkind IN ('m', 'v')
        WHERE d.refclassid = CAST('pg_class' AS regclass) AND d.refobjid <> rw.ev_class
      ),
      walk(root, upstream, path) AS (
        SELECT s.oid, e.upstream, ARRAY[s.oid, e.upstream]
        FROM selected s JOIN edges e ON e.dependent = s.oid
        UNION ALL
        SELECT w.root, e.upstream, w.path || e.upstream
        FROM walk w JOIN edges e ON e.dependent = w.upstream
        WHERE NOT e.upstream = ANY(w.path)
      ),
      ancestors AS (
        SELECT DISTINCT w.root, w.upstream
        FROM walk w JOIN pg_class c ON c.oid = w.upstream AND c.relkind = 'm'
      ),
      problems AS (
        SELECT 'Request must contain 1 to 25 views' AS reason
        WHERE (SELECT count(*) FROM requested) NOT BETWEEN 1 AND 25
        UNION ALL
        SELECT 'Duplicate view: ' || qualified_name FROM selected
        GROUP BY qualified_name HAVING count(*) > 1
        UNION ALL
        SELECT 'Not a materialized view: ' || qualified_name FROM selected
        WHERE relkind IS DISTINCT FROM 'm'
        UNION ALL
        SELECT 'Invalid schema/name entry' FROM selected
        WHERE (jsonb_typeof(entry->'schema') = 'string' AND schema_name <> ''
           AND jsonb_typeof(entry->'name') = 'string' AND view_name <> '') IS NOT TRUE
        UNION ALL
        SELECT 'Not populated: ' || qualified_name FROM selected
        WHERE relkind = 'm' AND NOT relispopulated
        UNION ALL
        SELECT 'No eligible UNIQUE index: ' || s.qualified_name FROM selected s
        WHERE s.relkind = 'm' AND NOT EXISTS (
          SELECT 1 FROM pg_index i WHERE i.indrelid = s.oid
            AND i.indisunique AND i.indisvalid AND i.indimmediate
            AND i.indpred IS NULL AND i.indexprs IS NULL AND i.indnatts > 0
        )
        UNION ALL
        SELECT 'Missing MAINTAIN privilege: ' || qualified_name FROM selected
        WHERE relkind = 'm' AND NOT has_table_privilege(oid, 'MAINTAIN')
        UNION ALL
        SELECT 'Missing upstream materialized view: ' || format('%I.%I', n.nspname, c.relname)
        FROM ancestors a JOIN pg_class c ON c.oid = a.upstream
        JOIN pg_namespace n ON n.oid = c.relnamespace
        WHERE NOT EXISTS (SELECT 1 FROM selected s WHERE s.oid = a.upstream)
      ),
      ordered AS (
        SELECT s.qualified_name,
               'REFRESH MATERIALIZED VIEW CONCURRENTLY ' || s.qualified_name || ';' AS refresh_sql,
               (SELECT count(*) FROM ancestors a WHERE a.root = s.oid) AS dependency_count
        FROM selected s WHERE s.relkind = 'm'
      )
      SELECT CAST(COALESCE((SELECT jsonb_agg(to_jsonb(o) ORDER BY dependency_count, qualified_name)
                       FROM ordered o), CAST('[]' AS jsonb)) AS text) AS plan,
             CAST(COALESCE((SELECT jsonb_agg(DISTINCT reason) FROM problems), CAST('[]' AS jsonb)) AS text) AS problems;

  - id: require_complete_plan
    type: io.kestra.plugin.core.execution.Assert
    description: Any failed precondition stops the entire plan before the first
      refresh; inspect plan.row.problems.
    conditions:
      - "{{ fromJson(outputs.plan.row.problems) | length == 0 }}"

  - id: apply
    type: io.kestra.plugin.core.flow.If
    condition: "{{ not inputs.dry_run }}"
    then:
      - id: refresh_in_order
        type: io.kestra.plugin.core.flow.Loop
        values: "{{ fromJson(outputs.plan.row.plan) }}"
        concurrencyLimit: 1
        tasks:
          - id: refresh_view
            type: io.kestra.plugin.jdbc.postgresql.Queries
            description: Each refresh commits before its dependants start. Readers can
              continue using the materialized view.
            url: "{{ secret('POSTGRES_JDBC_URL') }}"
            username: "{{ secret('POSTGRES_USERNAME') }}"
            password: "{{ secret('POSTGRES_PASSWORD') }}"
            transaction: true
            fetchType: FETCH
            parameters:
              view_name: "{{ fromJson(item.value).qualified_name }}"
            sql: |
              SET LOCAL lock_timeout = '5s';
              SET LOCAL statement_timeout = '15min';
              {{ fromJson(item.value).refresh_sql }}
              SELECT :view_name AS refreshed_view, clock_timestamp() AS completed_at;

triggers:
  - id: hourly
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"
    disabled: true
    inputs:
      dry_run: false