Postgres health check

-- ============================================================================
--  Postgres health check - Metad Labs - v1.0 (2026-09)
--  https://metadlabs.com/tools/postgres-health-check
--
--  One read-only SELECT. It reads system catalogs and statistics views only:
--  no writes, no extensions created, no settings changed. Safe on production.
--
--  Run it (psql shown; any SQL client works, e.g. pgAdmin, DBeaver, the
--  Supabase / RDS / Neon SQL editors):
--
--    psql -X "$DATABASE_URL" -f pg-health-check.sql > health.txt
--
--  Tips
--  * Connect as a superuser, or a role granted pg_monitor, for complete
--    results. Other roles still work but see less of other users' sessions
--    and queries.
--  * Table and index checks cover the database you connect to. Run it once
--    per database you care about.
--  * Usage counters (index scans, query stats) accumulate since the last stats
--    reset. The stats_age row tells you how long that window is.
--  * Each row's check name is an anchor on the page above, with what it means
--    and how to fix it. Paste the output there for a readable report; it is
--    analyzed in your browser and never uploaded.
--
--  Supports PostgreSQL 12 and newer. MIT licensed.
-- ============================================================================

with
cfg as (
  select
    current_setting('server_version_num')::int / 10000 as major,
    split_part(current_setting('server_version'), ' ', 1) as version,
    current_setting('block_size')::int as bs,
    (select setting::bigint from pg_settings where name = 'autovacuum_freeze_max_age') as freeze_max_age,
    (select setting::int from pg_settings where name = 'max_connections') as max_conn,
    pg_is_in_recovery() as standby,
    now() - pg_postmaster_start_time() as uptime
),
db as (
  select d.*,
    extract(epoch from now() - coalesce(d.stats_reset, pg_postmaster_start_time())) / 86400.0 as days
  from pg_stat_database d
  where d.datname = current_database()
),
setting as (
  select name, setting, unit,
    case when unit in ('B', 'kB', '8kB', 'MB') and setting ~ '^[0-9]+$'
         then pg_size_pretty(setting::bigint * case unit when 'kB' then 1024 when '8kB' then 8192 when 'MB' then 1048576 else 1 end)
         else setting || coalesce(unit, '') end as pretty
  from pg_settings
),

-- ---------------------------------------------------------------- overview --
c_server_version as (
  select 1 as n, 'server_version'::text as chk,
    case when eol < current_date then 'critical'
         when eol < current_date + 365 then 'warning'
         else 'ok' end as sev,
    'PostgreSQL ' || version as obj,
    case when eol < current_date
           then 'major version ' || major || ' reached end of life in ' || to_char(eol, 'Mon YYYY') || ' and gets no more security or bug fixes. Plan the upgrade now.'
         when eol < current_date + 365
           then 'community support for version ' || major || ' ends in ' || to_char(eol, 'Mon YYYY') || '. Start planning the upgrade.'
         when known
           then 'supported until ' || to_char(eol, 'Mon YYYY') || '. Keep up with minor releases for security fixes.'
         else 'a current major version. Keep up with minor releases for security fixes.' end as finding,
    0::numeric as ord
  from (
    select cfg.*, v.eol is not null as known,
      coalesce(v.eol, case when cfg.major < 12 then date '2000-01-01' else date '2099-01-01' end) as eol
    from cfg
    left join (values (12, date '2024-11-14'), (13, date '2025-11-13'), (14, date '2026-11-12'),
                      (15, date '2027-11-11'), (16, date '2028-11-09'), (17, date '2029-11-08'),
                      (18, date '2030-11-14'), (19, date '2031-11-13')) v(major, eol)
      on v.major = cfg.major
  ) x
),
c_overview as (
  select 2 as n, 'overview'::text as chk, 'info'::text as sev, current_database()::text as obj,
    'database size ' || pg_size_pretty(pg_database_size(current_database()))
      || '; server up ' || justify_interval(date_trunc('minute', cfg.uptime))
      || case when cfg.standby then '; this is a read replica: write-side checks are limited' else '' end as finding,
    0::numeric as ord
  from cfg
),
c_stats_age as (
  select 3 as n, 'stats_age'::text as chk,
    case when stats_reset > now() - interval '7 days' then 'warning' else 'info' end as sev,
    current_database()::text as obj,
    case when stats_reset is null
           then 'usage counters have never been reset, so they span the life of the database (or since its last crash recovery)'
         else 'usage counters were last reset on ' || to_char(stats_reset, 'YYYY-MM-DD') || ' (' || round(days) || ' days ago)'
           || case when stats_reset > now() - interval '7 days'
                   then '. That is too short a window to trust the unused-index and query results.' else '' end
    end as finding,
    0::numeric as ord
  from db
),

-- ------------------------------------------------------ pg_stat_statements --
pgss_loaded as (
  select exists (select 1 from setting where name = 'pg_stat_statements.max') as loaded
),
pgss_src as (
  select n.nspname, e.extversion,
    exists (select 1 from pg_attribute a
            where a.attrelid = to_regclass(format('%I.pg_stat_statements', n.nspname))
              and a.attname = 'total_exec_time') as exec_cols
  from pg_extension e
  join pg_namespace n on n.oid = e.extnamespace
  cross join pgss_loaded
  where e.extname = 'pg_stat_statements' and pgss_loaded.loaded
),
pgss as (
  select x.*
  from pgss_src s,
  lateral xmltable('/table/row' passing query_to_xml(format($q$
      select queryid::text as queryid, calls, total_ms, mean_ms, rows, pct, query,
             rank_total <= 10 as is_top, eligible and rank_mean <= 10 as is_slow
      from (
        select *,
          row_number() over (order by total_ms desc) as rank_total,
          row_number() over (partition by eligible order by mean_ms desc) as rank_mean
        from (
          select queryid, calls, rows,
            round(%1$s::numeric, 1) as total_ms,
            round(%2$s::numeric, 1) as mean_ms,
            coalesce(round((100 * %1$s / nullif(sum(%1$s) over (), 0))::numeric, 1), 0) as pct,
            left(regexp_replace(query, '[[:space:][:cntrl:]]+', ' ', 'g'), 90) as query,
            calls >= 10 and query !~* '^\s*(vacuum|analyze|create|alter|drop|copy|refresh|reindex|cluster|checkpoint|truncate|grant|revoke|comment|lock)\M' as eligible
          from %3$I.pg_stat_statements
          where dbid = (select oid from pg_database where datname = current_database())
        ) s
      ) r
      where rank_total <= 10 or (eligible and rank_mean <= 10)
    $q$,
    case when s.exec_cols then 'total_exec_time' else 'total_time' end,
    case when s.exec_cols then 'mean_exec_time' else 'mean_time' end,
    s.nspname), false, false, '')
    columns queryid text, calls numeric, total_ms numeric, mean_ms numeric, rows numeric,
            pct numeric, query text, is_top boolean, is_slow boolean) x
),
c_pgss as (
  select 4 as n, 'pg_stat_statements'::text as chk,
    case when s.nspname is not null then 'ok' else 'warning' end as sev,
    'pg_stat_statements'::text as obj,
    case when s.nspname is not null
           then 'enabled (version ' || s.extversion || '), so the top_queries and slow_queries checks below are based on real query stats'
         when l.loaded
           then 'loaded but not installed in this database. Run CREATE EXTENSION pg_stat_statements; (no restart needed), then re-run this script for the query checks.'
         else 'not enabled, so the query checks were skipped. It is the single most useful Postgres performance tool: add it to shared_preload_libraries, restart, then CREATE EXTENSION pg_stat_statements;'
    end as finding,
    0::numeric as ord
  from pgss_loaded l
  left join pgss_src s on true
),
c_top_queries as (
  select 28 as n, 'top_queries'::text as chk, 'info'::text as sev,
    'query ' || queryid || ': ' || query as obj,
    round(pct, 1) || '% of total query time; ' || to_char(calls, 'FM999,999,999,999,990') || ' calls; '
      || to_char(mean_ms, 'FM999,999,990.0') || ' ms avg; '
      || to_char(round(rows / nullif(calls, 0), 1), 'FM999,999,990.0') || ' rows/call' as finding,
    pct as ord
  from pgss where is_top
),
c_slow_queries as (
  select 29 as n, 'slow_queries'::text as chk,
    case when mean_ms >= 1000 then 'warning' else 'info' end as sev,
    'query ' || queryid || ': ' || query as obj,
    to_char(mean_ms, 'FM999,999,990.0') || ' ms per call on average; '
      || to_char(calls, 'FM999,999,999,999,990') || ' calls; '
      || to_char(round(rows / nullif(calls, 0), 1), 'FM999,999,990.0') || ' rows/call' as finding,
    mean_ms as ord
  from pgss where is_slow and mean_ms >= 250
),

-- ------------------------------------------------ vacuum and wraparound --
xid as (
  select datname, age(datfrozenxid) as xid_age, mxid_age(datminmxid) as mxid_age,
    100.0 * greatest(age(datfrozenxid), mxid_age(datminmxid)) / 2147483648 as pct
  from pg_database
),
c_xid_wraparound as (
  select 5 as n, 'xid_wraparound'::text as chk,
    case when pct >= 50 then 'critical' else 'warning' end as sev,
    datname::text as obj,
    case when xid_age >= mxid_age
           then 'transaction ID age ' || to_char(xid_age, 'FM999,999,999,990')
         else 'multixact ID age ' || to_char(mxid_age, 'FM999,999,999,990') end
      || ': ' || round(pct) || '% of the way to the forced wraparound shutdown. Find what is stopping vacuum from freezing (long transactions, abandoned replication slots or prepared transactions) and fix it.' as finding,
    pct as ord
  from xid where pct >= 25
  union all
  select 5, 'xid_wraparound', 'ok', 'all databases',
    'oldest is ' || datname || ' at ' || round(pct, 1) || '% of the wraparound limit', 0
  from (select * from xid order by pct desc limit 1) x
  where not exists (select 1 from xid where pct >= 25)
),
frz as (
  select format('%I.%I', n.nspname, c.relname) as tbl,
    greatest(age(c.relfrozenxid), age(t.relfrozenxid)) as xid_age, c.relpersistence
  from pg_class c
  join pg_namespace n on n.oid = c.relnamespace
  left join pg_class t on t.oid = c.reltoastrelid
  where c.relkind in ('r', 'm')
),
c_table_freeze_age as (
  (select 6 as n, 'table_freeze_age'::text as chk,
    case when xid_age >= 1073741824 then 'critical' else 'warning' end as sev,
    tbl as obj,
    'transaction ID age ' || to_char(xid_age, 'FM999,999,999,990') || ', '
      || round(xid_age::numeric / cfg.freeze_max_age, 1) || 'x autovacuum_freeze_max_age. '
      || case when relpersistence = 't'
              then 'Temporary tables are never vacuumed by autovacuum: end the session that owns it.'
              else 'Autovacuum should have frozen it by now. Check for blockers, or run VACUUM (FREEZE, VERBOSE) on it.' end as finding,
    xid_age::numeric as ord
  from frz, cfg
  where xid_age > 1.5 * cfg.freeze_max_age
  order by xid_age desc limit 10)
  union all
  select 6, 'table_freeze_age', 'ok', current_database(),
    'no table is overdue for freezing (all under 1.5x autovacuum_freeze_max_age)', 0
  from cfg
  where not exists (select 1 from frz where xid_age > 1.5 * cfg.freeze_max_age)
),
long_tx as (
  select a.pid, a.usename, a.application_name, a.state, a.query,
    now() - a.xact_start as tx_age, now() - a.state_change as state_age
  from pg_stat_activity a
  where a.pid <> pg_backend_pid()
    and a.backend_type = 'client backend'
    and a.xact_start is not null
    and (a.xact_start < now() - interval '15 minutes'
         or (a.state like 'idle in transaction%' and a.state_change < now() - interval '5 minutes'))
),
c_long_transactions as (
  (select 7 as n, 'long_transactions'::text as chk,
    case when tx_age > interval '1 hour' then 'critical' else 'warning' end as sev,
    'pid ' || pid || ' (' || coalesce(usename::text, '?') || coalesce('/' || nullif(application_name, ''), '') || ')' as obj,
    coalesce(state, 'unknown state') || '; transaction open ' || date_trunc('second', tx_age)
      || case when state like 'idle in transaction%' then ', idle for ' || date_trunc('second', state_age) else '' end
      || '; ' || coalesce(left(regexp_replace(query, '\s+', ' ', 'g'), 70), '') as finding,
    extract(epoch from tx_age)::numeric as ord
  from long_tx
  order by tx_age desc limit 10)
  union all
  select 7, 'long_transactions', 'ok', current_database(),
    'no transaction open longer than 15 minutes and nothing idle in transaction for over 5 minutes', 0
  where not exists (select 1 from long_tx)
),
c_prepared_transactions as (
  select 8 as n, 'prepared_transactions'::text as chk, 'critical'::text as sev,
    gid::text as obj,
    'prepared ' || date_trunc('second', now() - prepared) || ' ago in database ' || database
      || ' and never committed or rolled back. It blocks vacuum and holds its locks until you run COMMIT PREPARED or ROLLBACK PREPARED.' as finding,
    extract(epoch from now() - prepared)::numeric as ord
  from pg_prepared_xacts
  where prepared < now() - interval '5 minutes'
  union all
  select 8, 'prepared_transactions', 'ok', current_database(), 'no forgotten prepared transactions', 0
  where not exists (select 1 from pg_prepared_xacts where prepared < now() - interval '5 minutes')
),
slots as (
  select s.slot_name, s.slot_type, s.active, s.database,
    case when not cfg.standby then pg_wal_lsn_diff(pg_current_wal_lsn(), s.restart_lsn) end as retained,
    greatest(age(s.xmin), age(s.catalog_xmin)) as xmin_age
  from pg_replication_slots s, cfg
),
c_replication_slots as (
  select 9 as n, 'replication_slots'::text as chk,
    case when not active and (coalesce(retained, 0) >= 1073741824 or coalesce(xmin_age, 0) >= 100000000) then 'critical'
         else 'warning' end as sev,
    slot_name::text as obj,
    case when active then 'active ' else 'INACTIVE ' end || slot_type || ' slot'
      || coalesce(' holding ' || pg_size_pretty(retained::bigint) || ' of WAL', '')
      || coalesce(', xmin age ' || to_char(xmin_age, 'FM999,999,999,990'), '')
      || case when active then '. Its consumer is falling behind.'
              else '. Nothing is reading it, so WAL piles up until the disk fills, and it blocks vacuum. Drop it with pg_drop_replication_slot() if it is abandoned.' end as finding,
    coalesce(retained, 0)::numeric as ord
  from slots
  where not active or coalesce(retained, 0) >= 5368709120
  union all
  select 9, 'replication_slots', 'ok', current_database(),
    case when count(*) = 0 then 'no replication slots'
         else count(*) || ' replication slot(s), all active and keeping up' end, 0
  from slots
  having not bool_or(not active or coalesce(retained, 0) >= 5368709120) or count(*) = 0
),

-- ----------------------------------------------------- sessions and locks --
waits as (
  select a.pid, a.usename, a.wait_event, now() - a.query_start as waited,
    pg_blocking_pids(a.pid) as blockers, a.query
  from pg_stat_activity a
  where a.wait_event_type = 'Lock' and a.pid <> pg_backend_pid()
),
c_lock_waits as (
  (select 10 as n, 'lock_waits'::text as chk,
    case when waited > interval '30 seconds' then 'warning' else 'info' end as sev,
    'pid ' || pid || ' (' || coalesce(usename::text, '?') || ')' as obj,
    'waiting on a ' || wait_event || ' lock for ' || date_trunc('second', waited)
      || coalesce(', blocked by pid ' || nullif(array_to_string(blockers, ', '), ''), '')
      || '; ' || coalesce(left(regexp_replace(query, '\s+', ' ', 'g'), 70), '') as finding,
    extract(epoch from waited)::numeric as ord
  from waits
  order by waited desc limit 10)
  union all
  select 10, 'lock_waits', 'ok', current_database(), 'no sessions waiting on locks right now', 0
  where not exists (select 1 from waits)
),
conn as (
  select count(*) as total,
    count(*) filter (where state = 'active') as active,
    count(*) filter (where state = 'idle') as idle,
    count(*) filter (where state like 'idle in transaction%') as idle_tx
  from pg_stat_activity
  where backend_type = 'client backend'
),
c_connections as (
  select 11 as n, 'connections'::text as chk,
    case when 100.0 * total / max_conn >= 90 then 'critical'
         when 100.0 * total / max_conn >= 75 then 'warning'
         when idle >= 50 and idle > 5 * active then 'info'
         else 'ok' end as sev,
    'max_connections = ' || max_conn as obj,
    total || ' of ' || max_conn || ' connections in use (' || round(100.0 * total / max_conn) || '%): '
      || active || ' active, ' || idle || ' idle, ' || idle_tx || ' idle in transaction'
      || case when idle >= 50 and idle > 5 * active
              then '. Most are idle: a pooler such as PgBouncer, or smaller app-side pools, would cut memory use and headroom problems.'
              else '' end as finding,
    0::numeric as ord
  from conn, cfg
),

-- ---------------------------------------------------- tables and vacuum --
dead as (
  select format('%I.%I', s.schemaname, s.relname) as tbl, s.n_live_tup, s.n_dead_tup,
    100.0 * s.n_dead_tup / nullif(s.n_live_tup + s.n_dead_tup, 0) as pct,
    greatest(s.last_vacuum, s.last_autovacuum) as last_vac,
    coalesce(c.reloptions::text ~ 'autovacuum_enabled=(false|off)', false) as av_off
  from pg_stat_user_tables s
  join pg_class c on c.oid = s.relid
  where s.n_dead_tup >= 10000
    and s.n_dead_tup >= 0.2 * (s.n_live_tup + s.n_dead_tup)
),
c_autovacuum_lag as (
  (select 12 as n, 'autovacuum_lag'::text as chk, 'warning'::text as sev, tbl as obj,
    round(pct) || '% dead rows (' || to_char(n_dead_tup, 'FM999,999,999,990') || ' dead, '
      || to_char(n_live_tup, 'FM999,999,999,990') || ' live), last vacuum '
      || coalesce(to_char(last_vac, 'YYYY-MM-DD'), 'never')
      || case when av_off then '; autovacuum is DISABLED on this table' else '' end as finding,
    n_dead_tup::numeric as ord
  from dead
  order by n_dead_tup desc limit 10)
  union all
  select 12, 'autovacuum_lag', 'ok', current_database(),
    'autovacuum is keeping up: no table has more than 20% dead rows', 0
  where not exists (select 1 from dead)
),
stale as (
  select format('%I.%I', schemaname, relname) as tbl, n_live_tup, n_mod_since_analyze,
    greatest(last_analyze, last_autoanalyze) as last_an
  from pg_stat_user_tables
  where n_live_tup >= 10000
    and (greatest(last_analyze, last_autoanalyze) is null
         or n_mod_since_analyze > 0.5 * n_live_tup)
),
c_stale_statistics as (
  (select 13 as n, 'stale_statistics'::text as chk, 'warning'::text as sev, tbl as obj,
    case when last_an is null
           then 'never analyzed (' || to_char(n_live_tup, 'FM999,999,999,990') || ' rows), so the planner is guessing row counts. Run ANALYZE on it.'
         else to_char(n_mod_since_analyze, 'FM999,999,999,990') || ' rows changed since the last analyze on '
           || to_char(last_an, 'YYYY-MM-DD') || ' (' || round(100.0 * n_mod_since_analyze / n_live_tup) || '% of the table). Plans may be based on stale numbers.'
    end as finding,
    n_live_tup::numeric as ord
  from stale
  order by n_live_tup desc limit 10)
  union all
  select 13, 'stale_statistics', 'ok', current_database(), 'planner statistics look fresh on all tables over 10,000 rows', 0
  where not exists (select 1 from stale)
),
bloat_cols as (
  select c.oid, n.nspname, c.relname, c.relpages, c.reltuples,
    coalesce((regexp_match(array_to_string(c.reloptions, ','), 'fillfactor=(\d+)'))[1]::int, 100) as ff,
    sum((1 - coalesce(s.null_frac, 0)) * coalesce(s.avg_width, 0)) as data_width,
    coalesce(max(s.null_frac), 0) > 0 as has_nulls,
    count(*) as natts,
    count(s.attname) as nstats
  from pg_class c
  join pg_namespace n on n.oid = c.relnamespace
  join pg_attribute a on a.attrelid = c.oid and a.attnum > 0 and not a.attisdropped
  left join pg_stats s on s.schemaname = n.nspname and s.tablename = c.relname
                      and s.attname = a.attname and not s.inherited
  where c.relkind in ('r', 'm')
    and n.nspname not in ('pg_catalog', 'information_schema') and n.nspname !~ '^pg_toast'
    and c.relpages >= 1280
    and c.reltuples > 0
  group by c.oid, n.nspname, c.relname, c.relpages, c.reltuples, c.reloptions
),
bloat as (
  select format('%I.%I', nspname, relname) as tbl, relpages::bigint * bs as size,
    greatest(relpages - est_pages, 0)::bigint * bs as wasted,
    greatest(100.0 * (relpages - est_pages) / relpages, 0) as pct
  from (
    select b.*, cfg.bs,
      -- 23-byte tuple header + null bitmap and the row data, each padded to 8 bytes,
      -- plus a 4-byte line pointer; the page keeps a 24-byte header and fillfactor free.
      ceil(b.reltuples / least(291, greatest(1, floor((cfg.bs - 24) * b.ff / 100.0 / (
        ceil((23 + case when b.has_nulls then ceil(b.natts / 8.0) else 0 end) / 8.0) * 8
        + ceil(b.data_width / 8.0) * 8 + 4))))) as est_pages
    from bloat_cols b, cfg
    where b.natts = b.nstats
  ) e
),
c_table_bloat as (
  (select 14 as n, 'table_bloat'::text as chk,
    case when pct >= 50 and wasted >= 1073741824 then 'warning' else 'info' end as sev,
    tbl as obj,
    'about ' || round(pct) || '% bloat: ~' || pg_size_pretty(wasted) || ' of ' || pg_size_pretty(size)
      || ' is empty space (an estimate from planner statistics)' as finding,
    wasted::numeric as ord
  from bloat
  where pct >= 30 and wasted >= 104857600
  order by wasted desc limit 10)
  union all
  select 14, 'table_bloat', 'ok', current_database(),
    'no table over 10 MB has more than an estimated 30% (100 MB) of bloat', 0
  where not exists (select 1 from bloat where pct >= 30 and wasted >= 104857600)
),

-- ------------------------------------------------------------- indexes --
idx as (
  select i.indexrelid, i.indrelid, ic.relam, i.indisunique, i.indisprimary, i.indisvalid,
    i.indnkeyatts, i.indnatts,
    string_to_array(i.indkey::text, ' ')::int[] as keys,
    string_to_array(i.indclass::text, ' ') as cls,
    string_to_array(i.indcollation::text, ' ') as colls,
    string_to_array(i.indoption::text, ' ') as opts,
    coalesce(pg_get_expr(i.indexprs, i.indrelid), '') as exprs,
    coalesce(pg_get_expr(i.indpred, i.indrelid), '') as pred,
    exists (select 1 from pg_constraint k where k.conindid = i.indexrelid) as has_constraint,
    format('%I.%I', n.nspname, ic.relname) as idx_name,
    format('%I.%I', n.nspname, tc.relname) as tbl_name,
    pg_relation_size(i.indexrelid) as size,
    regexp_replace(pg_get_indexdef(i.indexrelid), '^.* USING ', '') as def
  from pg_index i
  join pg_class ic on ic.oid = i.indexrelid
  join pg_class tc on tc.oid = i.indrelid
  join pg_namespace n on n.oid = ic.relnamespace
  where n.nspname not in ('pg_catalog', 'information_schema') and n.nspname !~ '^pg_toast'
),
unused as (
  select x.*, s.idx_scan
  from idx x
  join pg_stat_user_indexes s on s.indexrelid = x.indexrelid
  where s.idx_scan = 0 and x.indisvalid and not x.indisunique and not x.has_constraint
),
c_unused_indexes as (
  (select 15 as n, 'unused_indexes'::text as chk,
    case when size >= 10485760 and (select stats_reset is null or stats_reset < now() - interval '7 days' from db)
         then 'warning' else 'info' end as sev,
    idx_name as obj,
    'no scans since stats reset; ' || pg_size_pretty(size) || '; on ' || tbl_name || '; ' || def as finding,
    size::numeric as ord
  from unused
  order by size desc limit 10)
  union all
  select 15, 'unused_indexes', 'info', '(total)',
    count(*) || ' unused indexes in all, ' || pg_size_pretty(sum(size)) || ' combined (only the 10 largest are listed)', -1
  from unused
  having count(*) > 10
  union all
  select 15, 'unused_indexes', 'ok', current_database(),
    'every non-unique index has been used since stats reset', 0
  where not exists (select 1 from unused)
),
dups as (
  select tbl_name,
    array_agg(idx_name order by has_constraint desc, indisunique desc, indexrelid) as names,
    sum(size) - max(size) as saved,
    min(def) as def
  from idx
  where indisvalid
  group by indrelid, tbl_name, relam, keys, indnkeyatts, cls, colls, opts, exprs, pred
  having count(*) > 1
),
c_duplicate_indexes as (
  select 16 as n, 'duplicate_indexes'::text as chk, 'warning'::text as sev, tbl_name as obj,
    'identical indexes ' || def || ': keep ' || names[1] || ', drop ' || array_to_string(names[2:], ', ')
      || ' (saves ~' || pg_size_pretty(saved) || ')' as finding,
    saved::numeric as ord
  from dups
  union all
  select 16, 'duplicate_indexes', 'ok', current_database(), 'no duplicate indexes', 0
  where not exists (select 1 from dups)
),
redundant as (
  select l.idx_name, l.def, l.size, r.idx_name as by_name, r.def as by_def, l.tbl_name
  from idx l
  join idx r on r.indrelid = l.indrelid and r.indexrelid <> l.indexrelid and r.relam = l.relam
  where l.relam = (select oid from pg_am where amname = 'btree')
    and l.indisvalid and r.indisvalid
    and not l.indisunique and not l.has_constraint
    and l.exprs = '' and r.exprs = '' and l.pred = '' and r.pred = ''
    and l.indnatts = l.indnkeyatts
    and l.indnkeyatts < r.indnkeyatts
    and r.keys[1:l.indnkeyatts] = l.keys
    and r.cls[1:l.indnkeyatts] = l.cls
    and r.colls[1:l.indnkeyatts] = l.colls
    and r.opts[1:l.indnkeyatts] = l.opts
),
c_redundant_indexes as (
  (select 17 as n, 'redundant_indexes'::text as chk, 'info'::text as sev, idx_name as obj,
    def || ' is a leading prefix of ' || by_name || ' ' || by_def || ', which can serve the same queries. Dropping it saves '
      || pg_size_pretty(size) || ' and speeds up writes.' as finding,
    size::numeric as ord
  from redundant
  order by size desc limit 10)
  union all
  select 17, 'redundant_indexes', 'ok', current_database(), 'no index is a plain prefix of another', 0
  where not exists (select 1 from redundant)
),
c_invalid_indexes as (
  select 18 as n, 'invalid_indexes'::text as chk, 'warning'::text as sev, idx_name as obj,
    'INVALID on ' || tbl_name || ' (' || pg_size_pretty(size) || '), left by a failed or still-running CREATE INDEX CONCURRENTLY. Writes still maintain it, but queries never use it. Drop it and build it again.' as finding,
    size::numeric as ord
  from idx where not indisvalid
  union all
  select 18, 'invalid_indexes', 'ok', current_database(), 'no invalid indexes', 0
  where not exists (select 1 from idx where not indisvalid)
),
fk as (
  select k.conname, k.conrelid, k.confrelid,
    format('%I.%I', n.nspname, c.relname) as tbl,
    format('%I.%I', fn.nspname, f.relname) as parent,
    (select string_agg(quote_ident(a.attname), ', ' order by u.ord)
       from unnest(k.conkey) with ordinality u(attnum, ord)
       join pg_attribute a on a.attrelid = k.conrelid and a.attnum = u.attnum) as cols,
    coalesce(nullif(pg_relation_size(k.conrelid), 0),
             (select sum(pg_relation_size(p.relid)) from pg_partition_tree(k.conrelid) p)) as size
  from pg_constraint k
  join pg_class c on c.oid = k.conrelid
  join pg_namespace n on n.oid = c.relnamespace
  join pg_class f on f.oid = k.confrelid
  join pg_namespace fn on fn.oid = f.relnamespace
  where k.contype = 'f' and k.conparentid = 0
    and not exists (
      select 1 from pg_index i
      where i.indrelid = k.conrelid
        and i.indpred is null
        and array_length(k.conkey, 1) <= i.indnkeyatts
        and (string_to_array(i.indkey::text, ' ')::int2[])[1:array_length(k.conkey, 1)] @> k.conkey
        and (string_to_array(i.indkey::text, ' ')::int2[])[1:array_length(k.conkey, 1)] <@ k.conkey
    )
),
c_missing_fk_indexes as (
  (select 19 as n, 'missing_fk_indexes'::text as chk,
    case when size >= 52428800 then 'warning' else 'info' end as sev,
    tbl || ' (' || cols || ')' as obj,
    'foreign key ' || quote_ident(conname) || ' to ' || parent || ' has no index on ' || tbl || ' ('
      || pg_size_pretty(size) || '). Every delete or key update on ' || parent || ' scans it, and joins on it cannot use an index.' as finding,
    size::numeric as ord
  from fk
  where size >= 1048576
  order by size desc limit 10)
  union all
  select 19, 'missing_fk_indexes', 'ok', current_database(), 'every foreign key on a table over 1 MB has a matching index', 0
  where not exists (select 1 from fk where size >= 1048576)
),
seqs as (
  select format('%I.%I', schemaname, relname) as tbl, seq_scan, seq_tup_read, coalesce(idx_scan, 0) as idx_scan,
    pg_relation_size(relid) as size, seq_tup_read / seq_scan as avg_rows
  from pg_stat_user_tables
  where seq_scan >= 100
    and seq_tup_read / seq_scan >= 10000
    and pg_relation_size(relid) >= 10485760
),
c_sequential_scans as (
  (select 20 as n, 'sequential_scans'::text as chk,
    case when seq_scan > idx_scan then 'warning' else 'info' end as sev,
    tbl as obj,
    to_char(seq_scan, 'FM999,999,999,990') || ' sequential scans reading ~' || to_char(avg_rows, 'FM999,999,999,990')
      || ' rows each, vs ' || to_char(idx_scan, 'FM999,999,999,990') || ' index scans (' || pg_size_pretty(size)
      || '). Some query filtering this table is probably missing an index.' as finding,
    seq_tup_read::numeric as ord
  from seqs
  order by seq_tup_read desc limit 10)
  union all
  select 20, 'sequential_scans', 'ok', current_database(), 'no table over 10 MB is being read by repeated large sequential scans', 0
  where not exists (select 1 from seqs)
),
seq_all as (
  -- Built from catalogs by OID (like the pg_sequences view) so a role without
  -- access to some schemas still gets a result instead of an error.
  select c.oid, format('%I.%I', n.nspname, c.relname) as seq, ps.seqmax,
    has_sequence_privilege(c.oid, 'SELECT, USAGE') as readable
  from pg_sequence ps
  join pg_class c on c.oid = ps.seqrelid
  join pg_namespace n on n.oid = c.relnamespace
  where ps.seqincrement > 0
),
seq_use as (
  select s.seq, lv.last_value, col.col, col.typ,
    least(s.seqmax, coalesce(col.type_max, s.seqmax)) as cap
  from seq_all s
  cross join lateral (select case when s.readable then pg_sequence_last_value(s.oid) end as last_value) lv
  left join lateral (
    select format('%I.%I', ns.nspname, t.relname) || '.' || quote_ident(a.attname) as col,
      format_type(a.atttypid, a.atttypmod) as typ,
      case a.atttypid when 'int2'::regtype then 32767 when 'int4'::regtype then 2147483647 end as type_max
    from pg_depend d
    join pg_attribute a on a.attrelid = d.refobjid and a.attnum = d.refobjsubid
    join pg_class t on t.oid = d.refobjid
    join pg_namespace ns on ns.oid = t.relnamespace
    where d.classid = 'pg_class'::regclass and d.refclassid = 'pg_class'::regclass
      and d.objid = s.oid and d.deptype in ('a', 'i')
    limit 1
  ) col on true
  where lv.last_value is not null
),
c_sequence_exhaustion as (
  select 21 as n, 'sequence_exhaustion'::text as chk,
    case when 100.0 * last_value / cap >= 75 then 'critical' else 'warning' end as sev,
    coalesce(col, seq) as obj,
    round(100.0 * last_value / cap, 1) || '% used (' || to_char(last_value, 'FM999,999,999,999,999,990') || ' of '
      || to_char(cap, 'FM999,999,999,999,999,990') || coalesce(', column is ' || typ, '')
      || '). Inserts fail when it runs out: migrate the column to bigint before then.' as finding,
    (100.0 * last_value / cap)::numeric as ord
  from seq_use
  where 100.0 * last_value / cap >= 50
  union all
  select 21, 'sequence_exhaustion',
    case when hidden > 0 then 'info' else 'ok' end, current_database(),
    coalesce('highest is ' || (select coalesce(col, seq) || ' at ' || round(100.0 * last_value / cap, 1) || '% of its limit'
                              from seq_use order by 1.0 * last_value / cap desc limit 1) || '. ',
             case when hidden > 0 then '' else 'no sequences in use' end)
      || case when hidden > 0
              then hidden || ' sequence(s) could not be read by this role: run as the owner or a superuser to check them'
              else '' end, 0
  from (select count(*) filter (where not readable) as hidden from seq_all) h
  where not exists (select 1 from seq_use where 100.0 * last_value / cap >= 50)
),
c_tables_without_pk as (
  (select 26 as n, 'tables_without_pk'::text as chk, 'info'::text as sev,
    format('%I.%I', n.nspname, c.relname) as obj,
    'no primary key (' || pg_size_pretty(pg_total_relation_size(c.oid)) || '). Logical replication, pg_repack and most ORMs need one, and duplicate rows can creep in.' as finding,
    pg_total_relation_size(c.oid)::numeric as ord
  from pg_class c
  join pg_namespace n on n.oid = c.relnamespace
  where c.relkind in ('r', 'p') and not c.relispartition
    and n.nspname not in ('pg_catalog', 'information_schema') and n.nspname !~ '^pg_toast'
    and not exists (select 1 from pg_constraint k where k.conrelid = c.oid and k.contype = 'p')
    and not exists (select 1 from pg_depend d where d.classid = 'pg_class'::regclass and d.objid = c.oid and d.deptype = 'e')
    and pg_total_relation_size(c.oid) >= 1048576
  order by pg_total_relation_size(c.oid) desc limit 10)
  union all
  select 26, 'tables_without_pk', 'ok', current_database(), 'every table over 1 MB has a primary key', 0
  where not exists (
    select 1 from pg_class c
    join pg_namespace n on n.oid = c.relnamespace
    where c.relkind in ('r', 'p') and not c.relispartition
      and n.nspname not in ('pg_catalog', 'information_schema') and n.nspname !~ '^pg_toast'
      and not exists (select 1 from pg_constraint k where k.conrelid = c.oid and k.contype = 'p')
      and not exists (select 1 from pg_depend d where d.classid = 'pg_class'::regclass and d.objid = c.oid and d.deptype = 'e')
      and pg_total_relation_size(c.oid) >= 1048576)
),

-- ------------------------------------------------------ database activity --
c_cache_hit_ratio as (
  select 22 as n, 'cache_hit_ratio'::text as chk,
    case when blks_hit + blks_read < 1000000 then 'info'
         when 100.0 * blks_hit / (blks_hit + blks_read) < 95 then 'warning'
         when 100.0 * blks_hit / (blks_hit + blks_read) < 99 then 'info'
         else 'ok' end as sev,
    current_database()::text as obj,
    case when blks_hit + blks_read < 1000000 then 'not enough reads yet to judge'
         else round(100.0 * blks_hit / (blks_hit + blks_read), 2) || '% of block reads served from shared_buffers ('
           || (select pretty from setting where name = 'shared_buffers') || ')'
           || case when 100.0 * blks_hit / (blks_hit + blks_read) < 99
                   then '. The working set may not fit in memory, or large scans are churning the cache.' else '' end
    end as finding,
    0::numeric as ord
  from db
),
c_temp_files as (
  select 23 as n, 'temp_files'::text as chk,
    case when temp_bytes / greatest(days, 1) >= 1073741824 then 'warning'
         when temp_files > 0 then 'info' else 'ok' end as sev,
    current_database()::text as obj,
    case when temp_files = 0 then 'no queries have spilled to temporary files'
         else to_char(temp_files, 'FM999,999,999,990') || ' temp files, ' || pg_size_pretty(temp_bytes)
           || ' written (~' || pg_size_pretty((temp_bytes / greatest(days, 1))::bigint) || '/day). Sorts and hashes that exceed work_mem ('
           || (select pretty from setting where name = 'work_mem') || ') spill to disk.' end as finding,
    0::numeric as ord
  from db
),
ckpt as (
  select x.timed, x.requested
  from xmltable('/table/row' passing query_to_xml(
    case when (select major from cfg) >= 17
         then 'select num_timed as timed, num_requested as requested from pg_stat_checkpointer'
         else 'select checkpoints_timed as timed, checkpoints_req as requested from pg_stat_bgwriter' end,
    false, false, '') columns timed numeric, requested numeric) x
),
c_checkpoints as (
  select 24 as n, 'checkpoints'::text as chk,
    case when timed + requested >= 10 and requested > 0.2 * (timed + requested) then 'warning' else 'ok' end as sev,
    'max_wal_size = ' || (select pretty from setting where name = 'max_wal_size') as obj,
    to_char(requested, 'FM999,999,990') || ' of ' || to_char(timed + requested, 'FM999,999,990')
      || ' checkpoints were forced by WAL volume rather than the timer'
      || case when timed + requested >= 10 and requested > 0.2 * (timed + requested)
              then '. Raising max_wal_size spreads checkpoint I/O out and cuts full-page writes.' else '' end as finding,
    0::numeric as ord
  from ckpt
),
c_rollbacks as (
  select 25 as n, 'rollbacks'::text as chk,
    case when xact_commit + xact_rollback < 10000 then 'ok'
         when 100.0 * xact_rollback / (xact_commit + xact_rollback) >= 10 then 'warning'
         when 100.0 * xact_rollback / (xact_commit + xact_rollback) >= 2 then 'info'
         else 'ok' end as sev,
    current_database()::text as obj,
    round(100.0 * xact_rollback / nullif(xact_commit + xact_rollback, 0), 1) || '% of transactions rolled back ('
      || to_char(xact_rollback, 'FM999,999,999,990') || ')'
      || case when deadlocks > 0 then '; ' || deadlocks || ' deadlocks' else '; no deadlocks' end
      || case when xact_commit + xact_rollback >= 10000 and 100.0 * xact_rollback / (xact_commit + xact_rollback) >= 2
              then '. A high rate usually means application errors or retries; check the Postgres log.' else '' end as finding,
    0::numeric as ord
  from db
),

-- ------------------------------------------------------------- settings --
flags as (
  select * from (values
    ('autovacuum', 'critical', 'off', 'autovacuum is off. Dead rows pile up and the database heads for wraparound. Turn it back on.'),
    ('fsync', 'critical', 'off', 'fsync is off: a crash can corrupt the database. Only acceptable for throwaway data.'),
    ('full_page_writes', 'critical', 'off', 'full_page_writes is off: a crash can leave torn pages. Turn it back on unless your storage guarantees atomic 8 kB writes.'),
    ('shared_buffers', 'warning', '16384', 'shared_buffers is still the 128 MB default. A common starting point is about 25% of RAM.'),
    ('work_mem', 'info', '4096', 'work_mem is the 4 MB default. Sorts and hashes bigger than that spill to disk (see temp_files).'),
    ('maintenance_work_mem', 'info', '65536', 'maintenance_work_mem is the 64 MB default, which makes vacuum and index builds slower. 512 MB to 1 GB is common.'),
    ('effective_cache_size', 'info', '524288', 'effective_cache_size is the 4 GB default. Set it to roughly 50-75% of RAM so the planner knows how much the OS caches.'),
    ('random_page_cost', 'info', '4', 'random_page_cost is 4, which assumes spinning disks. On SSD or cloud storage, 1.1 lets the planner pick index scans more often.'),
    ('track_io_timing', 'info', 'off', 'track_io_timing is off, so EXPLAIN and pg_stat_statements cannot show time spent on I/O. The overhead is small on modern hardware.'),
    ('idle_in_transaction_session_timeout', 'info', '0', 'idle_in_transaction_session_timeout is 0: a client that forgets to commit can block vacuum indefinitely. 5-15 minutes is a common safety net.'),
    ('log_min_duration_statement', 'info', '-1', 'slow queries are not logged. Setting log_min_duration_statement to 500-1000 ms gives you a record of them.')
  ) f(name, sev, bad, msg)
),
c_settings as (
  select 27 as n, 'settings'::text as chk, f.sev, f.name as obj, f.msg as finding, 0::numeric as ord
  from flags f join setting s on s.name = f.name and s.setting = f.bad
  union all
  select 27, 'settings', case when s.setting::int > 1000 then 'warning' else 'info' end, 'max_connections',
    'max_connections is ' || s.setting || '. Each connection costs memory and adds contention; a pooler such as PgBouncer serves more clients with far fewer backends.', 0
  from setting s where s.name = 'max_connections' and s.setting::int > 500
  union all
  select 27, 'settings', 'ok', current_database(), 'no red flags in the core settings checked', 0
  where not exists (select 1 from flags f join setting s on s.name = f.name and s.setting = f.bad)
    and not exists (select 1 from setting s where s.name = 'max_connections' and s.setting::int > 500)
),
c_largest_tables as (
  (select 30 as n, 'largest_tables'::text as chk, 'info'::text as sev,
    format('%I.%I', n.nspname, c.relname) as obj,
    pg_size_pretty(pg_total_relation_size(c.oid)) || ' total: table ' || pg_size_pretty(pg_relation_size(c.oid))
      || ', indexes ' || pg_size_pretty(pg_indexes_size(c.oid))
      || ', TOAST ' || pg_size_pretty(greatest(pg_total_relation_size(c.oid) - pg_relation_size(c.oid) - pg_indexes_size(c.oid), 0))
      || case when c.reltuples > 0 then '; ~' || to_char(c.reltuples, 'FM999,999,999,999,990') || ' rows' else '' end as finding,
    pg_total_relation_size(c.oid)::numeric as ord
  from pg_class c
  join pg_namespace n on n.oid = c.relnamespace
  where c.relkind in ('r', 'm')
    and n.nspname not in ('pg_catalog', 'information_schema') and n.nspname !~ '^pg_toast'
  order by pg_total_relation_size(c.oid) desc limit 5)
),

results as (
  select * from c_server_version
  union all select * from c_overview
  union all select * from c_stats_age
  union all select * from c_pgss
  union all select * from c_xid_wraparound
  union all select * from c_table_freeze_age
  union all select * from c_long_transactions
  union all select * from c_prepared_transactions
  union all select * from c_replication_slots
  union all select * from c_lock_waits
  union all select * from c_connections
  union all select * from c_autovacuum_lag
  union all select * from c_stale_statistics
  union all select * from c_table_bloat
  union all select * from c_unused_indexes
  union all select * from c_duplicate_indexes
  union all select * from c_redundant_indexes
  union all select * from c_invalid_indexes
  union all select * from c_missing_fk_indexes
  union all select * from c_sequential_scans
  union all select * from c_sequence_exhaustion
  union all select * from c_cache_hit_ratio
  union all select * from c_temp_files
  union all select * from c_checkpoints
  union all select * from c_rollbacks
  union all select * from c_tables_without_pk
  union all select * from c_settings
  union all select * from c_top_queries
  union all select * from c_slow_queries
  union all select * from c_largest_tables
)
select
  sev as severity,
  chk as "check",
  -- One line per row, and no "|" so the output survives copy and paste.
  translate(regexp_replace(obj, '\s+', ' ', 'g'), '|', bar) as object,
  translate(regexp_replace(finding, '\s+', ' ', 'g'), '|', bar) as finding
from results,
  -- chr(166) is a broken bar. Plain ASCII elsewhere so any server encoding works.
  (select case when getdatabaseencoding() = 'UTF8' then chr(166) else '/' end as bar) b
order by case sev when 'critical' then 1 when 'warning' then 2 when 'info' then 3 else 4 end, n, ord desc, obj;