-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia
-- Licensed under the Apache License, Version 2.0
-- https://www.apache.org/licenses/LICENSE-2.0
--
-- pg_flight_recorder install bundle for pgfr_analyze
--
-- Self-contained single-file install. Paste into the Supabase SQL editor
-- (or any client that does not process psql metacommands) and run.
-- Wrapped in BEGIN/COMMIT so a partial failure rolls back cleanly.

BEGIN;

-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

-- pg_flight_recorder: pgfr_analyze module install script
--
-- Requires pgfr_record module installed first.
-- Uses psql `\ir` (include-relative); paths resolve against this file's
-- directory, so the script works from any working directory.
--
-- Files:
--   01_core_metrics.sql           schema, modification_rate, hot_update_ratio,
--                                 anomaly_report, summary_report, compare,
--                                 xmin_horizon_history, current_xmin_horizon_holder
--   02_ring_readers.sql           recent_waits_current, recent_activity_current,
--                                 recent_locks_current, wait_summary, statement_compare
--   03_activity_storms_regressions.sql  activity_at, detect_query_storms,
--                                 _diagnose_regression_causes, detect_regressions
--   04_reports.sql                performance_report, check_alerts, report (overloads)
--   05_capacity.sql               preflight_check, quarterly_review,
--                                 capacity_summary, capacity_report
--   06_table_analysis.sql         table_compare, table_hotspots
--   07_index_analysis.sql         unused_indexes, index_efficiency
--   08_config.sql                 config_changes, config_at, config_health_check,
--                                 db_role_config_* functions
--   09_incident_analysis.sql      what_happened_at, incident_timeline, blast_radius
--   10_v2_readers.sql             v2_time_range, statement/table/index_activity_v2,
--                                 ring v2 reader rewrites


-- ==========================================================
-- pgfr_analyze/sql/01_core_metrics.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pgfr_record.config WHERE key = 'schema_version') THEN
        RAISE EXCEPTION 'Flight Recorder core not installed. Run pgfr_record/install.sql first.';
    END IF;
END $$;

CREATE SCHEMA IF NOT EXISTS pgfr_analyze;

-- Calculates the rate of row modifications (INSERT/UPDATE/DELETE) over a time window
-- Returns modifications per second, or NULL if insufficient data
CREATE OR REPLACE FUNCTION pgfr_analyze.modification_rate(
    p_relid OID,
    p_window INTERVAL
)
RETURNS NUMERIC
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_first_snapshot RECORD;
    v_last_snapshot RECORD;
    v_delta_mods BIGINT;
    v_delta_seconds NUMERIC;
BEGIN
    -- Get earliest snapshot within window
    SELECT ts.n_tup_ins, ts.n_tup_upd, ts.n_tup_del, s.captured_at
    INTO v_first_snapshot
    FROM pgfr_record.table_snapshots ts
    JOIN pgfr_record.snapshots s ON s.id = ts.snapshot_id
    WHERE ts.relid = p_relid
      AND s.captured_at >= now() - p_window
    ORDER BY s.captured_at ASC
    LIMIT 1;

    -- Get latest snapshot
    SELECT ts.n_tup_ins, ts.n_tup_upd, ts.n_tup_del, s.captured_at
    INTO v_last_snapshot
    FROM pgfr_record.table_snapshots ts
    JOIN pgfr_record.snapshots s ON s.id = ts.snapshot_id
    WHERE ts.relid = p_relid
      AND s.captured_at >= now() - p_window
    ORDER BY s.captured_at DESC
    LIMIT 1;

    -- Need at least 2 distinct snapshots
    IF v_first_snapshot.captured_at IS NULL OR v_last_snapshot.captured_at IS NULL
       OR v_first_snapshot.captured_at = v_last_snapshot.captured_at THEN
        RETURN NULL;
    END IF;

    v_delta_mods := (COALESCE(v_last_snapshot.n_tup_ins, 0) + COALESCE(v_last_snapshot.n_tup_upd, 0) + COALESCE(v_last_snapshot.n_tup_del, 0))
                  - (COALESCE(v_first_snapshot.n_tup_ins, 0) + COALESCE(v_first_snapshot.n_tup_upd, 0) + COALESCE(v_first_snapshot.n_tup_del, 0));
    v_delta_seconds := EXTRACT(EPOCH FROM (v_last_snapshot.captured_at - v_first_snapshot.captured_at));

    IF v_delta_seconds <= 0 THEN
        RETURN NULL;
    END IF;

    RETURN ROUND(v_delta_mods::numeric / v_delta_seconds, 4);
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.modification_rate(OID, INTERVAL) IS 'Returns row modification rate (modifications/second) for a table over a time window';

-- Calculates the HOT (Heap-Only Tuple) update ratio for a table
-- Higher ratio indicates more efficient updates that don't require index maintenance
-- Returns percentage (0-100), or NULL if no updates
CREATE OR REPLACE FUNCTION pgfr_analyze.hot_update_ratio(
    p_relid OID
)
RETURNS NUMERIC
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_n_tup_upd BIGINT;
    v_n_tup_hot_upd BIGINT;
BEGIN
    -- Get latest snapshot for this table
    SELECT ts.n_tup_upd, ts.n_tup_hot_upd
    INTO v_n_tup_upd, v_n_tup_hot_upd
    FROM pgfr_record.table_snapshots ts
    JOIN pgfr_record.snapshots s ON s.id = ts.snapshot_id
    WHERE ts.relid = p_relid
    ORDER BY s.captured_at DESC
    LIMIT 1;

    IF v_n_tup_upd IS NULL OR v_n_tup_upd = 0 THEN
        RETURN NULL;
    END IF;

    RETURN ROUND((COALESCE(v_n_tup_hot_upd, 0)::numeric / v_n_tup_upd) * 100, 2);
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.hot_update_ratio(OID) IS 'Returns HOT update percentage (0-100) for a table based on latest snapshot';

-- Analyzes database metrics within a time window and reports detected anomalies (checkpoints, buffer pressure, lock contention, etc.)
-- Returns anomalies with severity levels and actionable remediation recommendations
CREATE OR REPLACE FUNCTION pgfr_analyze.anomaly_report(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TABLE(
    anomaly_type        TEXT,
    severity            TEXT,
    description         TEXT,
    metric_value        TEXT,
    threshold           TEXT,
    recommendation      TEXT
)
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_cmp RECORD;
    v_wait_pct NUMERIC;
    v_lock_count INTEGER;
    v_max_block_duration INTERVAL;
    v_datfrozenxid_age INTEGER;
    v_datminmxid_age INTEGER;
    v_table_xid_rec RECORD;
    v_table_mxid_rec RECORD;
    v_freeze_max_age BIGINT;
    v_warning_threshold BIGINT;
    v_critical_threshold BIGINT;
    v_mxid_freeze_max_age BIGINT;
    v_mxid_warning_threshold BIGINT;
    v_mxid_critical_threshold BIGINT;
    v_xid_warning_ratio NUMERIC;
    v_xid_critical_ratio NUMERIC;
    v_mxid_warning_ratio NUMERIC;
    v_mxid_critical_ratio NUMERIC;
    v_row RECORD;
BEGIN
    -- XID / MultiXID wraparound thresholds are a configurable fraction of the
    -- corresponding autovacuum_*_freeze_max_age GUC. Ratios are read from
    -- pgfr_record.config so operators can tune without patching analyze code —
    -- defaults are conservative (0.5 warning, 0.8 critical) but 0.5 may be too
    -- high for warning on busy clusters; see config keys xid_warning_ratio etc.
    -- Both GUCs are always defined, so current_setting() never misses. See:
    --   https://postgres.ai/docs/postgres-howtos/performance-optimization/monitoring/how-to-monitor-transaction-id-wraparound-risks
    v_xid_warning_ratio  := coalesce(pgfr_record._get_config('xid_warning_ratio',  '0.5')::numeric, 0.5);
    v_xid_critical_ratio := coalesce(pgfr_record._get_config('xid_critical_ratio', '0.8')::numeric, 0.8);
    v_mxid_warning_ratio  := coalesce(pgfr_record._get_config('mxid_warning_ratio',  '0.5')::numeric, 0.5);
    v_mxid_critical_ratio := coalesce(pgfr_record._get_config('mxid_critical_ratio', '0.8')::numeric, 0.8);
    v_freeze_max_age      := current_setting('autovacuum_freeze_max_age')::int8;
    v_warning_threshold   := (v_freeze_max_age  * v_xid_warning_ratio )::bigint;
    v_critical_threshold  := (v_freeze_max_age  * v_xid_critical_ratio)::bigint;
    v_mxid_freeze_max_age       := current_setting('autovacuum_multixact_freeze_max_age')::int8;
    v_mxid_warning_threshold    := (v_mxid_freeze_max_age * v_mxid_warning_ratio )::bigint;
    v_mxid_critical_threshold   := (v_mxid_freeze_max_age * v_mxid_critical_ratio)::bigint;

    SELECT * INTO v_cmp FROM pgfr_analyze.compare(p_start_time, p_end_time);
    IF v_cmp.checkpoint_occurred THEN
        anomaly_type := 'CHECKPOINT_DURING_WINDOW';
        severity := CASE
            WHEN v_cmp.ckpt_write_time_ms > 30000 THEN 'high'
            WHEN v_cmp.ckpt_write_time_ms > 10000 THEN 'medium'
            ELSE 'low'
        END;
        description := 'A checkpoint occurred during this time window';
        metric_value := format('write_time: %s ms, sync_time: %s ms',
                              round(v_cmp.ckpt_write_time_ms::numeric, 1),
                              round(v_cmp.ckpt_sync_time_ms::numeric, 1));
        threshold := 'Any checkpoint';
        recommendation := 'Consider increasing max_wal_size or scheduling heavy writes after checkpoint_timeout';
        RETURN NEXT;
    END IF;
    IF v_cmp.ckpt_requested_delta > 0 THEN
        anomaly_type := 'FORCED_CHECKPOINT';
        severity := 'high';
        description := 'WAL exceeded max_wal_size, forcing checkpoint';
        metric_value := format('%s forced checkpoints', v_cmp.ckpt_requested_delta);
        threshold := 'ckpt_requested_delta > 0';
        recommendation := 'Increase max_wal_size to prevent mid-batch checkpoints';
        RETURN NEXT;
    END IF;
    IF COALESCE(v_cmp.bgw_buffers_backend_delta, 0) > 0 THEN
        anomaly_type := 'BUFFER_PRESSURE';
        severity := CASE
            WHEN v_cmp.bgw_buffers_backend_delta > 1000 THEN 'high'
            WHEN v_cmp.bgw_buffers_backend_delta > 100 THEN 'medium'
            ELSE 'low'
        END;
        description := 'Backends forced to write buffers directly (shared_buffers exhaustion)';
        metric_value := format('%s backend buffer writes', v_cmp.bgw_buffers_backend_delta);
        threshold := 'bgw_buffers_backend_delta > 0';
        recommendation := 'Increase shared_buffers, reduce concurrent writers, or use faster storage';
        RETURN NEXT;
    END IF;
    IF COALESCE(v_cmp.bgw_buffers_backend_fsync_delta, 0) > 0 THEN
        anomaly_type := 'BACKEND_FSYNC';
        severity := 'high';
        description := 'Backends forced to perform fsync (severe I/O bottleneck)';
        metric_value := format('%s backend fsyncs', v_cmp.bgw_buffers_backend_fsync_delta);
        threshold := 'bgw_buffers_backend_fsync_delta > 0';
        recommendation := 'Urgent: increase shared_buffers, reduce write load, or upgrade storage';
        RETURN NEXT;
    END IF;
    IF COALESCE(v_cmp.temp_files_delta, 0) > 0 THEN
        anomaly_type := 'TEMP_FILE_SPILLS';
        severity := CASE
            WHEN v_cmp.temp_bytes_delta > 1073741824 THEN 'high'
            WHEN v_cmp.temp_bytes_delta > 104857600 THEN 'medium'
            ELSE 'low'
        END;
        description := 'Queries spilling to temp files (work_mem exhaustion)';
        metric_value := format('%s temp files, %s written',
                              v_cmp.temp_files_delta, v_cmp.temp_bytes_pretty);
        threshold := 'temp_files_delta > 0';
        recommendation := 'Increase work_mem for affected sessions or globally';
        RETURN NEXT;
    END IF;
    -- query both legacy ring and v2 ring tables; dedup by taking max across both
    SELECT count(DISTINCT blocked_pid), max(blocked_duration)
    INTO v_lock_count, v_max_block_duration
    FROM (
        -- legacy ring (pre-Phase 3)
        SELECT l.blocked_pid,
               l.blocked_duration
        FROM pgfr_record.lock_samples_ring l
        JOIN pgfr_record.samples_ring s ON s.slot_id = l.slot_id
        WHERE s.captured_at BETWEEN p_start_time AND p_end_time
          AND l.blocked_pid IS NOT NULL
        UNION ALL
        -- v2 ring
        SELECT ls.blocked_pid,
               ls.blocked_duration_s * interval '1 second' as blocked_duration
        FROM pgfr_record.lock_samples ls
        WHERE pgfr_record.epoch() + ls.sample_ts * interval '1 second'
              BETWEEN p_start_time AND p_end_time
    ) combined;
    IF v_lock_count > 0 THEN
        anomaly_type := 'LOCK_CONTENTION';
        severity := CASE
            WHEN v_max_block_duration > interval '30 seconds' THEN 'high'
            WHEN v_max_block_duration > interval '5 seconds' THEN 'medium'
            ELSE 'low'
        END;
        description := 'Lock contention detected';
        metric_value := format('%s blocked sessions, max duration: %s',
                              v_lock_count, v_max_block_duration);
        threshold := 'Any lock contention';
        recommendation := 'Check recent_locks for blocking queries; consider shorter transactions';
        RETURN NEXT;
    END IF;

    -- xmin horizon anomalies. Fires when the data or catalog xmin horizon
    -- crosses the same fraction-of-autovacuum_freeze_max_age thresholds that
    -- XID_WRAPAROUND_RISK uses, but emitted FIRST so cause precedes the
    -- wraparound symptom. Holder details come from snapshots.xmin_horizon_detail
    -- (JSONB; populated by the collector). Data and catalog fire independently:
    -- both can trigger when a logical slot pins xmin and catalog_xmin together.
    DECLARE
        v_xmin RECORD;
        v_xmin_rec TEXT;
        v_xmin_src TEXT;
    BEGIN
        SELECT xmin_data_horizon_age, slot_catalog_xmin_age, xmin_horizon_detail
          INTO v_xmin
        FROM pgfr_record.snapshots
        WHERE captured_at BETWEEN p_start_time AND p_end_time
          AND (xmin_data_horizon_age IS NOT NULL OR slot_catalog_xmin_age IS NOT NULL)
        ORDER BY captured_at DESC
        LIMIT 1;

        IF v_xmin.xmin_horizon_detail IS NOT NULL THEN
            v_xmin_src := v_xmin.xmin_horizon_detail->>'source';
            v_xmin_rec := CASE v_xmin_src
                WHEN 'activity' THEN
                    CASE
                        WHEN v_xmin.xmin_horizon_detail->'holder'->>'backend_type' = 'autovacuum worker' THEN
                            format('long autovacuum (pid=%s) holding xmin — check pg_stat_progress_vacuum',
                                v_xmin.xmin_horizon_detail->'holder'->>'pid')
                        WHEN v_xmin.xmin_horizon_detail->'holder'->>'state' LIKE 'idle in transaction%' THEN
                            format('PID %s idle in transaction for %ss (user=%s, db=%s) — pg_terminate_backend(%s); cancel does nothing on an idle backend. Prevent recurrence: set idle_in_transaction_session_timeout (or transaction_timeout on PG17+).',
                                v_xmin.xmin_horizon_detail->'holder'->>'pid',
                                v_xmin.xmin_horizon_detail->'holder'->>'xact_age_seconds',
                                v_xmin.xmin_horizon_detail->'holder'->>'usename',
                                v_xmin.xmin_horizon_detail->'holder'->>'datname',
                                v_xmin.xmin_horizon_detail->'holder'->>'pid')
                        WHEN v_xmin.xmin_horizon_detail->'holder'->>'state' = 'active' THEN
                            -- Active txn pinning xmin is just as bad as idle-in-txn — autovacuum
                            -- still can't freeze. pg_cancel_backend ends the current query (and
                            -- the implicit txn for autocommit, releasing xmin) but if the query
                            -- runs inside an explicit BEGIN the txn stays open and xmin will
                            -- typically reappear; escalate to pg_terminate_backend in that case.
                            format('PID %s active for %ss holding xmin (user=%s, db=%s, query: %s) — pg_cancel_backend(%s); escalate to pg_terminate_backend(%s) if backend_xmin reappears (txn was inside explicit BEGIN). Prevent recurrence: set statement_timeout (and transaction_timeout on PG17+).',
                                v_xmin.xmin_horizon_detail->'holder'->>'pid',
                                v_xmin.xmin_horizon_detail->'holder'->>'query_age_seconds',
                                v_xmin.xmin_horizon_detail->'holder'->>'usename',
                                v_xmin.xmin_horizon_detail->'holder'->>'datname',
                                v_xmin.xmin_horizon_detail->'holder'->>'query_preview',
                                v_xmin.xmin_horizon_detail->'holder'->>'pid',
                                v_xmin.xmin_horizon_detail->'holder'->>'pid')
                        ELSE
                            format('PID %s holding xmin (state=%s, backend_type=%s) — investigate via pg_stat_activity; pg_terminate_backend(%s) if no other path releases backend_xmin',
                                v_xmin.xmin_horizon_detail->'holder'->>'pid',
                                v_xmin.xmin_horizon_detail->'holder'->>'state',
                                v_xmin.xmin_horizon_detail->'holder'->>'backend_type',
                                v_xmin.xmin_horizon_detail->'holder'->>'pid')
                    END
                WHEN 'slot' THEN
                    format('investigate slot ''%s'' (type=%s); advance or DROP REPLICATION SLOT once subscriber state is confirmed',
                        v_xmin.xmin_horizon_detail->'holder'->>'slot_name',
                        v_xmin.xmin_horizon_detail->'holder'->>'slot_type')
                WHEN 'slot_catalog' THEN
                    format('logical slot ''%s'' pinning catalog_xmin — advance or DROP REPLICATION SLOT',
                        v_xmin.xmin_horizon_detail->'holder'->>'slot_name')
                WHEN 'prepared' THEN
                    format('ROLLBACK PREPARED ''%s'' (owner=%s, database=%s, prepared_at=%s)',
                        v_xmin.xmin_horizon_detail->'holder'->>'gid',
                        v_xmin.xmin_horizon_detail->'holder'->>'owner',
                        v_xmin.xmin_horizon_detail->'holder'->>'database',
                        v_xmin.xmin_horizon_detail->'holder'->>'prepared_at')
                WHEN 'replication' THEN
                    -- The walsender PID is on the primary; the actual cause is a
                    -- long-running query on the standby whose snapshot xmin is
                    -- being shipped back via hot_standby_feedback. Standby
                    -- workloads are usually autocommit reads, so pg_cancel_backend
                    -- on the standby is the right first action. Don't disable
                    -- hot_standby_feedback — that just trades xmin holdback for
                    -- "canceling statement due to conflict with recovery" failures
                    -- on the standby.
                    format('standby ''%s'' (addr=%s, slot=%s) is reporting backend_xmin via hot_standby_feedback — connect to that standby, find the long-running query in its pg_stat_activity, and pg_cancel_backend it there. Do not disable hot_standby_feedback (causes recovery-conflict cancellations on the standby).',
                        v_xmin.xmin_horizon_detail->'holder'->>'application_name',
                        v_xmin.xmin_horizon_detail->'holder'->>'client_addr',
                        v_xmin.xmin_horizon_detail->'holder'->>'slot_name')
                ELSE 'see snapshots.xmin_horizon_detail'
            END;
        END IF;

        IF v_xmin.xmin_data_horizon_age IS NOT NULL
           AND v_xmin.xmin_data_horizon_age > v_warning_threshold THEN
            anomaly_type := 'XMIN_HORIZON_STALL';
            severity := CASE WHEN v_xmin.xmin_data_horizon_age > v_critical_threshold
                             THEN 'critical' ELSE 'high' END;
            description := 'xmin horizon stalled — autovacuum cannot freeze user-table tuples';
            metric_value := format('xmin_data_horizon_age=%s (%s%% of autovacuum_freeze_max_age); source=%s',
                to_char(v_xmin.xmin_data_horizon_age, 'FM999,999,999'),
                round(v_xmin.xmin_data_horizon_age::numeric / v_freeze_max_age * 100, 1),
                COALESCE(v_xmin_src, 'unknown'));
            threshold := format('xmin_data_horizon_age > %s%% of autovacuum_freeze_max_age',
                round(v_xid_warning_ratio * 100, 0));
            recommendation := COALESCE(v_xmin_rec, 'see snapshots.xmin_horizon_detail');
            RETURN NEXT;
        END IF;

        IF v_xmin.slot_catalog_xmin_age IS NOT NULL
           AND v_xmin.slot_catalog_xmin_age > v_warning_threshold THEN
            anomaly_type := 'CATALOG_XMIN_HORIZON_STALL';
            severity := CASE WHEN v_xmin.slot_catalog_xmin_age > v_critical_threshold
                             THEN 'critical' ELSE 'high' END;
            description := 'logical-replication catalog cleanup stalled by replication slot catalog_xmin';
            metric_value := format('slot_catalog_xmin_age=%s (%s%% of autovacuum_freeze_max_age)',
                to_char(v_xmin.slot_catalog_xmin_age, 'FM999,999,999'),
                round(v_xmin.slot_catalog_xmin_age::numeric / v_freeze_max_age * 100, 1));
            threshold := format('slot_catalog_xmin_age > %s%% of autovacuum_freeze_max_age',
                round(v_xid_warning_ratio * 100, 0));
            recommendation := COALESCE(
                CASE WHEN v_xmin_src IN ('slot','slot_catalog') THEN v_xmin_rec END,
                'investigate logical replication slots with non-null catalog_xmin (see pg_replication_slots)');
            RETURN NEXT;
        END IF;
    END;

    -- Database-level XID wraparound check
    SELECT datfrozenxid_age INTO v_datfrozenxid_age
    FROM pgfr_record.snapshots
    WHERE captured_at BETWEEN p_start_time AND p_end_time
      AND datfrozenxid_age IS NOT NULL
    ORDER BY captured_at DESC
    LIMIT 1;
    IF v_datfrozenxid_age IS NOT NULL AND v_datfrozenxid_age > v_warning_threshold THEN
        anomaly_type := 'XID_WRAPAROUND_RISK';
        severity := CASE
            WHEN v_datfrozenxid_age > v_critical_threshold THEN 'critical'
            ELSE 'high'
        END;
        description := 'Database approaching transaction ID wraparound';
        metric_value := format('XID age: %s (%s%% of autovacuum_freeze_max_age)',
                              to_char(v_datfrozenxid_age, 'FM999,999,999'),
                              round(v_datfrozenxid_age::numeric / v_freeze_max_age * 100, 1));
        threshold := format('datfrozenxid_age > %s (50%% of %s)',
                           to_char(v_warning_threshold, 'FM999,999,999'),
                           to_char(v_freeze_max_age, 'FM999,999,999'));
        recommendation := 'Run VACUUM FREEZE on large tables or enable more aggressive autovacuum';
        RETURN NEXT;
    END IF;
    -- Database-level MultiXID wraparound check. See:
    --   https://postgres.ai/docs/postgres-howtos/performance-optimization/monitoring/how-to-monitor-transaction-id-wraparound-risks
    SELECT datminmxid_age INTO v_datminmxid_age
    FROM pgfr_record.snapshots
    WHERE captured_at BETWEEN p_start_time AND p_end_time
      AND datminmxid_age IS NOT NULL
    ORDER BY captured_at DESC
    LIMIT 1;
    IF v_datminmxid_age IS NOT NULL AND v_datminmxid_age > v_mxid_warning_threshold THEN
        anomaly_type := 'MXID_WRAPAROUND_RISK';
        severity := CASE
            WHEN v_datminmxid_age > v_mxid_critical_threshold THEN 'critical'
            ELSE 'high'
        END;
        description := 'Database approaching MultiXact ID wraparound';
        metric_value := format('MXID age: %s (%s%% of autovacuum_multixact_freeze_max_age)',
                              to_char(v_datminmxid_age, 'FM999,999,999'),
                              round(v_datminmxid_age::numeric / v_mxid_freeze_max_age * 100, 1));
        threshold := format('datminmxid_age > %s (50%% of %s)',
                           to_char(v_mxid_warning_threshold, 'FM999,999,999'),
                           to_char(v_mxid_freeze_max_age, 'FM999,999,999'));
        recommendation := 'Investigate row-level locks (SELECT FOR SHARE/UPDATE), foreign-key-heavy updates, and unfrozen MultiXacts. Run VACUUM (FREEZE) on largest tables; autovacuum may be blocked by long transactions or idle-in-transaction sessions.';
        RETURN NEXT;
    END IF;
    -- Table-level XID wraparound check (find tables approaching their threshold)
    -- Each table may have its own autovacuum_freeze_max_age setting
    FOR v_table_xid_rec IN
        SELECT
            COALESCE(ts.schemaname, split_part(ts.relid::regclass::text, '.', 1)) AS schemaname,
            COALESCE(ts.relname, split_part(ts.relid::regclass::text, '.', 2)) AS relname,
            ts.relfrozenxid_age,
            COALESCE(
                (SELECT (regexp_match(opt, 'autovacuum_freeze_max_age=(\d+)'))[1]::bigint
                 FROM unnest(c.reloptions) opt
                 WHERE opt LIKE 'autovacuum_freeze_max_age=%'
                 LIMIT 1),
                v_freeze_max_age
            ) AS table_freeze_max_age
        FROM pgfr_record.table_snapshots ts
        LEFT JOIN pg_class c ON c.oid = ts.relid
        WHERE ts.snapshot_id = (
            SELECT id FROM pgfr_record.snapshots
            WHERE captured_at BETWEEN p_start_time AND p_end_time
            ORDER BY captured_at DESC
            LIMIT 1
        )
          AND ts.relfrozenxid_age IS NOT NULL
          AND COALESCE(c.relkind, 'r') <> 'p' -- exclude partitioned tables (relfrozenxid is always 0)
        ORDER BY ts.relfrozenxid_age::numeric / COALESCE(
            (SELECT (regexp_match(opt, 'autovacuum_freeze_max_age=(\d+)'))[1]::bigint
             FROM unnest(c.reloptions) opt
             WHERE opt LIKE 'autovacuum_freeze_max_age=%'
             LIMIT 1),
            v_freeze_max_age
        ) DESC
        LIMIT 5  -- Check top 5 tables by relative XID age
    LOOP
        IF v_table_xid_rec.relfrozenxid_age > (v_table_xid_rec.table_freeze_max_age * v_xid_warning_ratio)::bigint THEN
            anomaly_type := 'TABLE_XID_WRAPAROUND_RISK';
            severity := CASE
                WHEN v_table_xid_rec.relfrozenxid_age > (v_table_xid_rec.table_freeze_max_age * v_xid_critical_ratio)::bigint THEN 'critical'
                ELSE 'high'
            END;
            description := format('Table %s.%s approaching XID wraparound',
                                 v_table_xid_rec.schemaname, v_table_xid_rec.relname);
            metric_value := format('XID age: %s (%s%% of table autovacuum_freeze_max_age=%s)',
                                  to_char(v_table_xid_rec.relfrozenxid_age, 'FM999,999,999'),
                                  round(v_table_xid_rec.relfrozenxid_age::numeric / v_table_xid_rec.table_freeze_max_age * 100, 1),
                                  to_char(v_table_xid_rec.table_freeze_max_age, 'FM999,999,999'));
            threshold := format('relfrozenxid_age > %s (%s%% of %s)',
                               to_char((v_table_xid_rec.table_freeze_max_age * v_xid_warning_ratio)::bigint, 'FM999,999,999'),
                               round(v_xid_warning_ratio * 100),
                               to_char(v_table_xid_rec.table_freeze_max_age, 'FM999,999,999'));
            recommendation := format('Run VACUUM FREEZE on %s.%s',
                                    v_table_xid_rec.schemaname, v_table_xid_rec.relname);
            RETURN NEXT;
        END IF;
    END LOOP;

    -- Table-level MultiXID wraparound check. Each table may have its own
    -- autovacuum_multixact_freeze_max_age reloption override. See:
    --   https://postgres.ai/docs/postgres-howtos/performance-optimization/monitoring/how-to-monitor-transaction-id-wraparound-risks
    FOR v_table_mxid_rec IN
        SELECT
            COALESCE(ts.schemaname, split_part(ts.relid::regclass::text, '.', 1)) AS schemaname,
            COALESCE(ts.relname, split_part(ts.relid::regclass::text, '.', 2)) AS relname,
            ts.relminmxid_age,
            COALESCE(
                (SELECT (regexp_match(opt, 'autovacuum_multixact_freeze_max_age=(\d+)'))[1]::bigint
                 FROM unnest(c.reloptions) opt
                 WHERE opt LIKE 'autovacuum_multixact_freeze_max_age=%'
                 LIMIT 1),
                v_mxid_freeze_max_age
            ) AS table_mxid_freeze_max_age
        FROM pgfr_record.table_snapshots ts
        LEFT JOIN pg_class c ON c.oid = ts.relid
        WHERE ts.snapshot_id = (
            SELECT id FROM pgfr_record.snapshots
            WHERE captured_at BETWEEN p_start_time AND p_end_time
            ORDER BY captured_at DESC
            LIMIT 1
        )
          AND ts.relminmxid_age IS NOT NULL
          AND COALESCE(c.relkind, 'r') <> 'p' -- exclude partitioned tables (relminmxid is always 0)
        ORDER BY ts.relminmxid_age::numeric / COALESCE(
            (SELECT (regexp_match(opt, 'autovacuum_multixact_freeze_max_age=(\d+)'))[1]::bigint
             FROM unnest(c.reloptions) opt
             WHERE opt LIKE 'autovacuum_multixact_freeze_max_age=%'
             LIMIT 1),
            v_mxid_freeze_max_age
        ) DESC
        LIMIT 5
    LOOP
        IF v_table_mxid_rec.relminmxid_age > (v_table_mxid_rec.table_mxid_freeze_max_age * v_mxid_warning_ratio)::bigint THEN
            anomaly_type := 'TABLE_MXID_WRAPAROUND_RISK';
            severity := CASE
                WHEN v_table_mxid_rec.relminmxid_age > (v_table_mxid_rec.table_mxid_freeze_max_age * v_mxid_critical_ratio)::bigint THEN 'critical'
                ELSE 'high'
            END;
            description := format('Table %s.%s approaching MultiXact ID wraparound',
                                 v_table_mxid_rec.schemaname, v_table_mxid_rec.relname);
            metric_value := format('MXID age: %s (%s%% of table autovacuum_multixact_freeze_max_age=%s)',
                                  to_char(v_table_mxid_rec.relminmxid_age, 'FM999,999,999'),
                                  round(v_table_mxid_rec.relminmxid_age::numeric / v_table_mxid_rec.table_mxid_freeze_max_age * 100, 1),
                                  to_char(v_table_mxid_rec.table_mxid_freeze_max_age, 'FM999,999,999'));
            threshold := format('relminmxid_age > %s (%s%% of %s)',
                               to_char((v_table_mxid_rec.table_mxid_freeze_max_age * v_mxid_warning_ratio)::bigint, 'FM999,999,999'),
                               round(v_mxid_warning_ratio * 100),
                               to_char(v_table_mxid_rec.table_mxid_freeze_max_age, 'FM999,999,999'));
            recommendation := format('Run VACUUM FREEZE on %s.%s; check for row-level lock / FK contention generating unfrozen MultiXacts',
                                    v_table_mxid_rec.schemaname, v_table_mxid_rec.relname);
            RETURN NEXT;
        END IF;
    END LOOP;

    -- OID exhaustion detection
    -- OIDs are 32-bit unsigned integers (max ~4.3 billion)
    -- Unlike XIDs, OIDs are not recycled - they simply exhaust
    DECLARE
        v_max_catalog_oid BIGINT;
        v_large_object_count BIGINT;
        v_oid_max BIGINT := 4294967295;  -- 2^32 - 1
        v_oid_warning_threshold BIGINT := (v_oid_max * 0.75)::bigint;   -- 75% = ~3.22 billion
        v_oid_critical_threshold BIGINT := (v_oid_max * 0.90)::bigint;  -- 90% = ~3.87 billion
    BEGIN
        SELECT max_catalog_oid, large_object_count
        INTO v_max_catalog_oid, v_large_object_count
        FROM pgfr_record.snapshots
        WHERE captured_at BETWEEN p_start_time AND p_end_time
          AND max_catalog_oid IS NOT NULL
        ORDER BY captured_at DESC
        LIMIT 1;

        IF v_max_catalog_oid IS NOT NULL AND v_max_catalog_oid > v_oid_warning_threshold THEN
            anomaly_type := 'OID_EXHAUSTION_RISK';
            severity := CASE
                WHEN v_max_catalog_oid > v_oid_critical_threshold THEN 'critical'
                ELSE 'high'
            END;
            description := 'Database approaching OID exhaustion';
            metric_value := format('Max catalog OID: %s (%s%% of 4.3 billion), Large objects: %s',
                                  to_char(v_max_catalog_oid, 'FM999,999,999,999'),
                                  round(v_max_catalog_oid::numeric / v_oid_max * 100, 1),
                                  COALESCE(to_char(v_large_object_count, 'FM999,999,999'), 'N/A'));
            threshold := format('max_catalog_oid > %s (75%% of 4,294,967,295)',
                               to_char(v_oid_warning_threshold, 'FM999,999,999,999'));
            recommendation := 'OID exhaustion requires pg_dump/pg_restore to reset counter. Review lo_create() usage and large object cleanup.';
            RETURN NEXT;
        END IF;
    END;

    -- Idle-in-transaction detection
    FOR v_row IN
        SELECT pid, usename, application_name,
               EXTRACT(EPOCH FROM (now() - xact_start))/60 AS idle_minutes
        FROM pgfr_record.activity_samples_archive
        WHERE captured_at BETWEEN p_start_time AND p_end_time
          AND state = 'idle in transaction'
          AND xact_start IS NOT NULL
          AND now() - xact_start > interval '5 minutes'
        ORDER BY xact_start ASC
        LIMIT 5
    LOOP
        anomaly_type := 'IDLE_IN_TRANSACTION';
        severity := CASE WHEN v_row.idle_minutes > 60 THEN 'critical'
             WHEN v_row.idle_minutes > 15 THEN 'high'
             ELSE 'medium' END;
        description := format('Session %s (%s) idle in transaction for %s minutes',
               v_row.pid, v_row.usename, round(v_row.idle_minutes::numeric));
        metric_value := format('PID %s, %s minutes', v_row.pid, round(v_row.idle_minutes::numeric));
        threshold := '>5 minutes idle in transaction';
        recommendation := 'Investigate and terminate if stale. Blocks vacuum and holds locks.';
        RETURN NEXT;
    END LOOP;

    -- Dead tuple accumulation (bloat risk)
    FOR v_row IN
        SELECT COALESCE(ts.schemaname, split_part(ts.relid::regclass::text, '.', 1)) AS schemaname,
               COALESCE(ts.relname, split_part(ts.relid::regclass::text, '.', 2)) AS relname,
               ts.n_dead_tup, ts.n_live_tup,
               round(100.0 * ts.n_dead_tup / NULLIF(ts.n_dead_tup + ts.n_live_tup, 0), 1) AS dead_pct
        FROM pgfr_record.table_snapshots ts
        JOIN pgfr_record.snapshots s ON s.id = ts.snapshot_id
        WHERE s.captured_at = (SELECT MAX(s2.captured_at) FROM pgfr_record.snapshots s2
                               JOIN pgfr_record.table_snapshots ts2 ON ts2.snapshot_id = s2.id
                               WHERE s2.captured_at <= p_end_time)
          AND ts.n_dead_tup > 10000
          AND ts.n_dead_tup::float / NULLIF(ts.n_dead_tup + ts.n_live_tup, 0) > 0.1
        ORDER BY ts.n_dead_tup DESC
        LIMIT 5
    LOOP
        anomaly_type := 'DEAD_TUPLE_ACCUMULATION';
        severity := CASE WHEN v_row.dead_pct > 30 THEN 'high'
             ELSE 'medium' END;
        description := format('Table %s.%s has %s%% dead tuples (%s dead)',
               v_row.schemaname, v_row.relname, v_row.dead_pct, v_row.n_dead_tup);
        metric_value := format('%s%% dead tuples', v_row.dead_pct);
        threshold := '>10% dead tuples and >10000 dead rows';
        recommendation := 'Run VACUUM on this table. Check autovacuum settings.';
        RETURN NEXT;
    END LOOP;

    -- Vacuum starvation (dead tuples growing, vacuum not running)
    FOR v_row IN
        WITH recent AS (
            SELECT COALESCE(ts.schemaname, split_part(ts.relid::regclass::text, '.', 1)) AS schemaname,
                   COALESCE(ts.relname, split_part(ts.relid::regclass::text, '.', 2)) AS relname,
                   ts.relid, ts.n_dead_tup, ts.last_autovacuum,
                   s.captured_at,
                   LAG(ts.n_dead_tup) OVER (PARTITION BY ts.relid ORDER BY s.captured_at) AS prev_dead
            FROM pgfr_record.table_snapshots ts
            JOIN pgfr_record.snapshots s ON s.id = ts.snapshot_id
            WHERE s.captured_at BETWEEN p_start_time AND p_end_time
        )
        SELECT schemaname, relname, n_dead_tup, last_autovacuum,
               n_dead_tup - COALESCE(prev_dead, 0) AS dead_growth
        FROM recent
        WHERE captured_at = (SELECT MAX(captured_at) FROM recent)
          AND n_dead_tup > prev_dead + 1000
          AND (last_autovacuum IS NULL OR last_autovacuum < now() - interval '24 hours')
        ORDER BY n_dead_tup - COALESCE(prev_dead, 0) DESC
        LIMIT 3
    LOOP
        anomaly_type := 'VACUUM_STARVATION';
        severity := 'high';
        description := format('Table %s.%s: dead tuples growing (+%s) but no vacuum in 24h',
               v_row.schemaname, v_row.relname, v_row.dead_growth);
        metric_value := format('+%s dead tuples, last vacuum: %s', v_row.dead_growth,
               COALESCE(v_row.last_autovacuum::TEXT, 'never'));
        threshold := 'Dead tuples growing >1000 with no vacuum in 24h';
        recommendation := 'Check autovacuum_vacuum_threshold and autovacuum_vacuum_scale_factor.';
        RETURN NEXT;
    END LOOP;

    -- Connection leak detection (sessions open > 7 days)
    FOR v_row IN
        SELECT DISTINCT ON (pid) pid, usename, application_name, backend_start,
               EXTRACT(DAY FROM (now() - backend_start)) AS days_open
        FROM pgfr_record.activity_samples_archive
        WHERE captured_at BETWEEN p_start_time AND p_end_time
          AND backend_start IS NOT NULL
          AND backend_start < now() - interval '7 days'
        ORDER BY pid, backend_start
        LIMIT 5
    LOOP
        anomaly_type := 'CONNECTION_LEAK';
        severity := CASE WHEN v_row.days_open > 30 THEN 'high' ELSE 'medium' END;
        description := format('Session %s (%s/%s) open for %s days',
               v_row.pid, v_row.usename, v_row.application_name, round(v_row.days_open::numeric));
        metric_value := format('%s days', round(v_row.days_open::numeric));
        threshold := '>7 days session age';
        recommendation := 'Investigate if this is a connection leak. Consider connection pooling.';
        RETURN NEXT;
    END LOOP;

    -- Replication lag velocity (lag is growing)
    FOR v_row IN
        WITH lag_samples AS (
            SELECT r.application_name,
                   EXTRACT(EPOCH FROM r.replay_lag) AS lag_seconds,
                   s.captured_at,
                   ROW_NUMBER() OVER (PARTITION BY r.application_name ORDER BY s.captured_at) AS rn
            FROM pgfr_record.replication_snapshots r
            JOIN pgfr_record.snapshots s ON s.id = r.snapshot_id
            WHERE s.captured_at BETWEEN p_start_time AND p_end_time
              AND r.replay_lag IS NOT NULL
        ),
        lag_trend AS (
            SELECT application_name,
                   MAX(lag_seconds) - MIN(lag_seconds) AS lag_growth,
                   MAX(lag_seconds) AS current_lag,
                   COUNT(*) AS samples
            FROM lag_samples
            GROUP BY application_name
            HAVING COUNT(*) >= 3
        )
        SELECT * FROM lag_trend
        WHERE lag_growth > 60  -- Growing by more than 60 seconds
          AND current_lag > 30 -- And currently > 30 seconds behind
    LOOP
        anomaly_type := 'REPLICATION_LAG_GROWING';
        severity := CASE WHEN v_row.current_lag > 300 THEN 'critical'
             WHEN v_row.current_lag > 60 THEN 'high'
             ELSE 'medium' END;
        description := format('Replica %s: lag growing (+%ss), now %ss behind',
               v_row.application_name, round(v_row.lag_growth::numeric), round(v_row.current_lag::numeric));
        metric_value := format('+%ss growth, %ss current', round(v_row.lag_growth::numeric), round(v_row.current_lag::numeric));
        threshold := '>60s growth and >30s current lag';
        recommendation := 'Check replica capacity, network, and long-running queries on primary.';
        RETURN NEXT;
    END LOOP;

    RETURN;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.anomaly_report(TIMESTAMPTZ, TIMESTAMPTZ) IS
'Analyzes database metrics within a time window and reports detected anomalies (checkpoints, buffer pressure, lock contention, XID wraparound, bloat, replication lag) with severity levels and remediation recommendations.';

-- Generates a comprehensive performance report with metrics and interpretations for a specified time window
-- Aggregates data from compare, anomaly detection, wait events, and lock contention to provide human-readable insights
CREATE OR REPLACE FUNCTION pgfr_analyze.summary_report(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TABLE(
    section             TEXT,
    metric              TEXT,
    value               TEXT,
    interpretation      TEXT
)
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_cmp RECORD;
    v_sample_count INTEGER;
    v_anomaly_count INTEGER;
    v_top_wait RECORD;
    v_lock_summary RECORD;
BEGIN
    SELECT * INTO v_cmp FROM pgfr_analyze.compare(p_start_time, p_end_time);
    SELECT count(*) INTO v_sample_count
    FROM (
        SELECT 1 FROM pgfr_record.samples_ring
        WHERE captured_at BETWEEN p_start_time AND p_end_time
        UNION ALL
        SELECT 1 FROM pgfr_record.activity_samples
        WHERE pgfr_record.epoch() + sample_ts * interval '1 second'
              BETWEEN p_start_time AND p_end_time
    ) combined;
    SELECT count(*) INTO v_anomaly_count
    FROM pgfr_analyze.anomaly_report(p_start_time, p_end_time);
    section := 'OVERVIEW';
    metric := 'Time Window';
    value := format('%s to %s', p_start_time, p_end_time);
    interpretation := format('%s seconds elapsed', round(COALESCE(v_cmp.elapsed_seconds, 0), 1));
    RETURN NEXT;
    metric := 'Data Coverage';
    value := format('%s snapshots, %s samples',
                   (SELECT count(*) FROM pgfr_record.snapshots
                    WHERE captured_at BETWEEN p_start_time AND p_end_time),
                   v_sample_count);
    interpretation := CASE
        WHEN v_sample_count = 0 THEN 'WARNING: No sample data in window'
        ELSE 'OK'
    END;
    RETURN NEXT;
    metric := 'Anomalies Detected';
    value := v_anomaly_count::text;
    interpretation := CASE
        WHEN v_anomaly_count = 0 THEN 'No issues detected'
        WHEN v_anomaly_count <= 2 THEN 'Minor issues'
        ELSE 'Multiple issues - review anomaly_report()'
    END;
    RETURN NEXT;
    section := 'CHECKPOINT & WAL';
    metric := 'Checkpoint Occurred';
    value := COALESCE(v_cmp.checkpoint_occurred::text, 'unknown');
    interpretation := CASE
        WHEN v_cmp.checkpoint_occurred THEN
            format('Checkpoint write: %s ms', round(COALESCE(v_cmp.ckpt_write_time_ms, 0)::numeric, 1))
        ELSE 'No checkpoint during window'
    END;
    RETURN NEXT;
    metric := 'WAL Generated';
    value := COALESCE(v_cmp.wal_bytes_pretty, 'N/A');
    interpretation := format('%s MB/sec',
                            round((COALESCE(v_cmp.wal_bytes_delta, 0) / 1048576.0) / NULLIF(v_cmp.elapsed_seconds, 0), 2));
    RETURN NEXT;
    section := 'BUFFERS & I/O';
    metric := 'Buffers Allocated';
    value := COALESCE(v_cmp.bgw_buffers_alloc_delta::text, 'N/A');
    interpretation := 'New buffer allocations';
    RETURN NEXT;
    metric := 'Backend Buffer Writes';
    value := COALESCE(v_cmp.bgw_buffers_backend_delta::text, 'N/A');
    interpretation := CASE
        WHEN COALESCE(v_cmp.bgw_buffers_backend_delta, 0) > 0 THEN 'WARNING: Backends writing directly'
        ELSE 'OK'
    END;
    RETURN NEXT;
    metric := 'Temp File Spills';
    value := format('%s files, %s', COALESCE(v_cmp.temp_files_delta, 0), COALESCE(v_cmp.temp_bytes_pretty, '0 B'));
    interpretation := CASE
        WHEN COALESCE(v_cmp.temp_files_delta, 0) > 0 THEN 'Consider increasing work_mem'
        ELSE 'No temp file usage'
    END;
    RETURN NEXT;
    section := 'WAIT EVENTS';
    FOR v_top_wait IN
        SELECT wait_event_type || ':' || wait_event AS we, total_waiters, pct_of_samples
        FROM pgfr_analyze.wait_summary(p_start_time, p_end_time)
        LIMIT 5
    LOOP
        metric := v_top_wait.we;
        value := format('%s total waiters (%s%% of samples)',
                       v_top_wait.total_waiters, v_top_wait.pct_of_samples);
        interpretation := CASE
            WHEN v_top_wait.we LIKE 'Lock:%' THEN 'Lock contention'
            WHEN v_top_wait.we LIKE 'LWLock:Buffer%' THEN 'Buffer contention'
            WHEN v_top_wait.we LIKE 'LWLock:WAL%' THEN 'WAL contention'
            WHEN v_top_wait.we LIKE 'IO:%' THEN 'I/O bound'
            WHEN v_top_wait.we = 'Running:CPU' THEN 'CPU active (normal)'
            ELSE 'Review PostgreSQL docs'
        END;
        RETURN NEXT;
    END LOOP;
    section := 'LOCK CONTENTION';
    SELECT
        count(DISTINCT blocked_pid) AS blocked_count,
        max(blocked_duration) AS max_duration
    INTO v_lock_summary
    FROM (
        SELECT l.blocked_pid, l.blocked_duration
        FROM pgfr_record.lock_samples_ring l
        JOIN pgfr_record.samples_ring s ON s.slot_id = l.slot_id
        WHERE s.captured_at BETWEEN p_start_time AND p_end_time
          AND l.blocked_pid IS NOT NULL
        UNION ALL
        SELECT ls.blocked_pid,
               ls.blocked_duration_s * interval '1 second'
        FROM pgfr_record.lock_samples ls
        WHERE pgfr_record.epoch() + ls.sample_ts * interval '1 second'
              BETWEEN p_start_time AND p_end_time
    ) combined;
    metric := 'Blocked Sessions';
    value := COALESCE(v_lock_summary.blocked_count, 0)::text;
    interpretation := CASE
        WHEN COALESCE(v_lock_summary.blocked_count, 0) = 0 THEN 'No lock contention'
        ELSE format('Max blocked duration: %s', v_lock_summary.max_duration)
    END;
    RETURN NEXT;
    RETURN;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.summary_report(TIMESTAMPTZ, TIMESTAMPTZ) IS
'Generates a comprehensive performance report aggregating compare, anomaly detection, wait events, and lock contention data into human-readable metrics and interpretations.';

-- =============================================================================
-- ANALYSIS FUNCTIONS (full implementations, cross-schema reads from pgfr_record.*)
-- =============================================================================

-- Compares database metrics between two time points, returning checkpoint, WAL, buffer, and IO activity deltas
CREATE OR REPLACE FUNCTION pgfr_analyze.compare(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TABLE(
    start_snapshot_at       TIMESTAMPTZ,
    end_snapshot_at         TIMESTAMPTZ,
    elapsed_seconds         NUMERIC,
    checkpoint_occurred     BOOLEAN,
    ckpt_timed_delta        BIGINT,
    ckpt_requested_delta    BIGINT,
    ckpt_write_time_ms      NUMERIC,
    ckpt_sync_time_ms       NUMERIC,
    ckpt_buffers_delta      BIGINT,
    wal_bytes_delta         BIGINT,
    wal_bytes_pretty        TEXT,
    wal_write_time_ms       NUMERIC,
    wal_sync_time_ms        NUMERIC,
    bgw_buffers_clean_delta       BIGINT,
    bgw_buffers_alloc_delta       BIGINT,
    bgw_buffers_backend_delta     BIGINT,
    bgw_buffers_backend_fsync_delta BIGINT,
    slots_count             INTEGER,
    slots_max_retained_wal  BIGINT,
    slots_max_retained_pretty TEXT,
    io_ckpt_reads_delta           BIGINT,
    io_ckpt_read_time_ms          NUMERIC,
    io_ckpt_writes_delta          BIGINT,
    io_ckpt_write_time_ms         NUMERIC,
    io_ckpt_fsyncs_delta          BIGINT,
    io_ckpt_fsync_time_ms         NUMERIC,
    io_autovacuum_reads_delta     BIGINT,
    io_autovacuum_read_time_ms    NUMERIC,
    io_autovacuum_writes_delta    BIGINT,
    io_autovacuum_write_time_ms   NUMERIC,
    io_client_reads_delta         BIGINT,
    io_client_read_time_ms        NUMERIC,
    io_client_writes_delta        BIGINT,
    io_client_write_time_ms       NUMERIC,
    io_bgwriter_reads_delta       BIGINT,
    io_bgwriter_read_time_ms      NUMERIC,
    io_bgwriter_writes_delta      BIGINT,
    io_bgwriter_write_time_ms     NUMERIC,
    temp_files_delta              BIGINT,
    temp_bytes_delta              BIGINT,
    temp_bytes_pretty             TEXT
)
LANGUAGE sql STABLE AS $$
    WITH
    start_snap AS (
        SELECT * FROM pgfr_record.snapshots
        WHERE captured_at <= p_start_time
        ORDER BY captured_at DESC
        LIMIT 1
    ),
    end_snap AS (
        SELECT * FROM pgfr_record.snapshots
        WHERE captured_at >= p_end_time
        ORDER BY captured_at ASC
        LIMIT 1
    )
    SELECT
        s.captured_at,
        e.captured_at,
        EXTRACT(EPOCH FROM (e.captured_at - s.captured_at))::numeric,
        (s.checkpoint_time IS DISTINCT FROM e.checkpoint_time),
        e.ckpt_timed - s.ckpt_timed,
        e.ckpt_requested - s.ckpt_requested,
        (e.ckpt_write_time - s.ckpt_write_time)::numeric,
        (e.ckpt_sync_time - s.ckpt_sync_time)::numeric,
        e.ckpt_buffers - s.ckpt_buffers,
        e.wal_bytes - s.wal_bytes,
        pgfr_record._pretty_bytes(e.wal_bytes - s.wal_bytes),
        (e.wal_write_time - s.wal_write_time)::numeric,
        (e.wal_sync_time - s.wal_sync_time)::numeric,
        e.bgw_buffers_clean - s.bgw_buffers_clean,
        e.bgw_buffers_alloc - s.bgw_buffers_alloc,
        e.bgw_buffers_backend - s.bgw_buffers_backend,
        e.bgw_buffers_backend_fsync - s.bgw_buffers_backend_fsync,
        e.slots_count,
        e.slots_max_retained_wal,
        pgfr_record._pretty_bytes(e.slots_max_retained_wal),
        e.io_checkpointer_reads - s.io_checkpointer_reads,
        (e.io_checkpointer_read_time - s.io_checkpointer_read_time)::numeric,
        e.io_checkpointer_writes - s.io_checkpointer_writes,
        (e.io_checkpointer_write_time - s.io_checkpointer_write_time)::numeric,
        e.io_checkpointer_fsyncs - s.io_checkpointer_fsyncs,
        (e.io_checkpointer_fsync_time - s.io_checkpointer_fsync_time)::numeric,
        e.io_autovacuum_reads - s.io_autovacuum_reads,
        (e.io_autovacuum_read_time - s.io_autovacuum_read_time)::numeric,
        e.io_autovacuum_writes - s.io_autovacuum_writes,
        (e.io_autovacuum_write_time - s.io_autovacuum_write_time)::numeric,
        e.io_client_reads - s.io_client_reads,
        (e.io_client_read_time - s.io_client_read_time)::numeric,
        e.io_client_writes - s.io_client_writes,
        (e.io_client_write_time - s.io_client_write_time)::numeric,
        e.io_bgwriter_reads - s.io_bgwriter_reads,
        (e.io_bgwriter_read_time - s.io_bgwriter_read_time)::numeric,
        e.io_bgwriter_writes - s.io_bgwriter_writes,
        (e.io_bgwriter_write_time - s.io_bgwriter_write_time)::numeric,
        e.temp_files - s.temp_files,
        e.temp_bytes - s.temp_bytes,
        pgfr_record._pretty_bytes(e.temp_bytes - s.temp_bytes)
    FROM start_snap s, end_snap e
$$;
COMMENT ON FUNCTION pgfr_analyze.compare(TIMESTAMPTZ, TIMESTAMPTZ) IS 'Compares database metrics between two time points, returning checkpoint, WAL, buffer, and IO activity deltas.';

-- xmin horizon readers. Both read pgfr_record.snapshots.xmin_horizon_detail
-- (JSONB) directly; no sidecar joins. The collector populates the JSONB
-- with {source, age, holder} for the dominant holder per snapshot. See
-- REFERENCE.md "xmin horizon monitoring".

CREATE OR REPLACE FUNCTION pgfr_analyze.xmin_horizon_history(
    p_start TIMESTAMPTZ,
    p_end   TIMESTAMPTZ
)
RETURNS TABLE (
    captured_at           TIMESTAMPTZ,
    xmin_data_horizon_age BIGINT,
    slot_catalog_xmin_age BIGINT,
    xmin_any_horizon_age  BIGINT,
    source                TEXT,
    holder                JSONB
)
LANGUAGE sql STABLE AS $$
    SELECT s.captured_at,
           s.xmin_data_horizon_age,
           s.slot_catalog_xmin_age,
           s.xmin_any_horizon_age,
           s.xmin_horizon_detail->>'source' AS source,
           s.xmin_horizon_detail->'holder'  AS holder
    FROM pgfr_record.snapshots s
    WHERE s.captured_at BETWEEN p_start AND p_end
      AND (s.xmin_data_horizon_age IS NOT NULL OR s.slot_catalog_xmin_age IS NOT NULL)
    ORDER BY s.captured_at DESC;
$$;
COMMENT ON FUNCTION pgfr_analyze.xmin_horizon_history(TIMESTAMPTZ, TIMESTAMPTZ) IS
'Timeline of xmin horizon ages and dominant-holder JSONB detail across snapshots in a window.';

CREATE OR REPLACE FUNCTION pgfr_analyze.current_xmin_horizon_holder()
RETURNS TABLE (
    captured_at           TIMESTAMPTZ,
    xmin_data_horizon_age BIGINT,
    slot_catalog_xmin_age BIGINT,
    source                TEXT,
    holder                JSONB
)
LANGUAGE sql STABLE AS $$
    SELECT s.captured_at,
           s.xmin_data_horizon_age,
           s.slot_catalog_xmin_age,
           s.xmin_horizon_detail->>'source' AS source,
           s.xmin_horizon_detail->'holder'  AS holder
    FROM pgfr_record.snapshots s
    WHERE s.xmin_horizon_detail IS NOT NULL
    ORDER BY s.captured_at DESC
    LIMIT 1;
$$;
COMMENT ON FUNCTION pgfr_analyze.current_xmin_horizon_holder() IS
'Quick-answer reader: zero rows on a healthy cluster, otherwise one row for the dominant xmin holder at the latest snapshot.';

-- Retrieves recent wait event samples from the flight recorder ring buffer

-- ==========================================================
-- pgfr_analyze/sql/02_ring_readers.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

CREATE OR REPLACE FUNCTION pgfr_analyze.recent_waits_current()
RETURNS TABLE (
    captured_at TIMESTAMPTZ,
    backend_type TEXT,
    wait_event_type TEXT,
    wait_event TEXT,
    state TEXT,
    count INTEGER
)
LANGUAGE sql STABLE AS $$
    SELECT
        sr.captured_at,
        w.backend_type,
        w.wait_event_type,
        w.wait_event,
        w.state,
        w.count
    FROM pgfr_record.samples_ring sr
    JOIN pgfr_record.wait_samples_ring w ON w.slot_id = sr.slot_id
    WHERE sr.captured_at > now() - pgfr_record._get_ring_retention_interval()
      AND w.backend_type IS NOT NULL
    ORDER BY sr.captured_at DESC, w.count DESC;
$$;
COMMENT ON FUNCTION pgfr_analyze.recent_waits_current() IS 'Retrieves recent wait event samples from the flight recorder ring buffer.';

-- Retrieves recent backend session activity with query duration and wait event details
CREATE OR REPLACE FUNCTION pgfr_analyze.recent_activity_current()
RETURNS TABLE (
    captured_at TIMESTAMPTZ,
    pid INTEGER,
    usename TEXT,
    application_name TEXT,
    backend_type TEXT,
    state TEXT,
    wait_event_type TEXT,
    wait_event TEXT,
    query_start TIMESTAMPTZ,
    running_for INTERVAL,
    query_preview TEXT
)
LANGUAGE sql STABLE AS $$
    SELECT
        sr.captured_at,
        a.pid,
        a.usename,
        a.application_name,
        a.backend_type,
        a.state,
        a.wait_event_type,
        a.wait_event,
        a.query_start,
        sr.captured_at - a.query_start AS running_for,
        a.query_preview
    FROM pgfr_record.samples_ring sr
    JOIN pgfr_record.activity_samples_ring a ON a.slot_id = sr.slot_id
    WHERE sr.captured_at > now() - pgfr_record._get_ring_retention_interval()
      AND a.pid IS NOT NULL
    ORDER BY sr.captured_at DESC, a.query_start ASC;
$$;
COMMENT ON FUNCTION pgfr_analyze.recent_activity_current() IS 'Retrieves recent backend session activity with query duration and wait event details.';

-- Returns recent lock contention events showing which processes are blocked
CREATE OR REPLACE FUNCTION pgfr_analyze.recent_locks_current()
RETURNS TABLE (
    captured_at TIMESTAMPTZ,
    blocked_pid INTEGER,
    blocked_user TEXT,
    blocked_app TEXT,
    blocked_duration INTERVAL,
    blocking_pid INTEGER,
    blocking_user TEXT,
    blocking_app TEXT,
    lock_type TEXT,
    locked_relation TEXT,
    blocked_query_preview TEXT,
    blocking_query_preview TEXT
)
LANGUAGE sql STABLE AS $$
    SELECT
        sr.captured_at,
        l.blocked_pid,
        l.blocked_user,
        l.blocked_app,
        l.blocked_duration,
        l.blocking_pid,
        l.blocking_user,
        l.blocking_app,
        l.lock_type,
        COALESCE(l.locked_relation_oid::regclass::text, 'OID:' || l.locked_relation_oid::text) AS locked_relation,
        l.blocked_query_preview,
        l.blocking_query_preview
    FROM pgfr_record.samples_ring sr
    JOIN pgfr_record.lock_samples_ring l ON l.slot_id = sr.slot_id
    WHERE sr.captured_at > now() - pgfr_record._get_ring_retention_interval()
      AND l.blocked_pid IS NOT NULL
    ORDER BY sr.captured_at DESC, l.blocked_duration DESC;
$$;
COMMENT ON FUNCTION pgfr_analyze.recent_locks_current() IS 'Returns recent lock contention events showing which processes are blocked.';

-- Summarizes wait events within a time range, grouped by backend type and wait event
CREATE OR REPLACE FUNCTION pgfr_analyze.wait_summary(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TABLE(
    backend_type        TEXT,
    wait_event_type     TEXT,
    wait_event          TEXT,
    sample_count        BIGINT,
    total_waiters       BIGINT,
    avg_waiters         NUMERIC,
    max_waiters         INTEGER,
    pct_of_samples      NUMERIC
)
LANGUAGE sql STABLE AS $$
    WITH sample_range AS (
        SELECT slot_id, captured_at
        FROM pgfr_record.samples_ring
        WHERE captured_at BETWEEN p_start_time AND p_end_time
    ),
    total_samples AS (
        SELECT count(*) AS cnt FROM sample_range
    )
    SELECT
        w.backend_type,
        w.wait_event_type,
        w.wait_event,
        count(DISTINCT w.slot_id) AS sample_count,
        sum(w.count) AS total_waiters,
        round(avg(w.count), 2) AS avg_waiters,
        max(w.count) AS max_waiters,
        round(100.0 * count(DISTINCT w.slot_id) / NULLIF(t.cnt, 0), 1) AS pct_of_samples
    FROM pgfr_record.wait_samples_ring w
    JOIN sample_range sr ON sr.slot_id = w.slot_id
    CROSS JOIN total_samples t
    WHERE w.state NOT IN ('idle', 'idle in transaction')
    GROUP BY w.backend_type, w.wait_event_type, w.wait_event, t.cnt
    ORDER BY total_waiters DESC, sample_count DESC;
$$;
COMMENT ON FUNCTION pgfr_analyze.wait_summary(TIMESTAMPTZ, TIMESTAMPTZ) IS 'Summarizes wait events within a time range, grouped by backend type and wait event.';

-- Compares statement execution metrics between two snapshots, calculating performance deltas
CREATE OR REPLACE FUNCTION pgfr_analyze.statement_compare(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ,
    p_min_delta_ms DOUBLE PRECISION DEFAULT 100,
    p_limit INTEGER DEFAULT 25
)
RETURNS TABLE(
    queryid                     BIGINT,
    query_preview               TEXT,
    calls_start                 BIGINT,
    calls_end                   BIGINT,
    calls_delta                 BIGINT,
    total_exec_time_start_ms    DOUBLE PRECISION,
    total_exec_time_end_ms      DOUBLE PRECISION,
    total_exec_time_delta_ms    DOUBLE PRECISION,
    mean_exec_time_start_ms     DOUBLE PRECISION,
    mean_exec_time_end_ms       DOUBLE PRECISION,
    rows_delta                  BIGINT,
    shared_blks_hit_delta       BIGINT,
    shared_blks_read_delta      BIGINT,
    shared_blks_written_delta   BIGINT,
    temp_blks_read_delta        BIGINT,
    temp_blks_written_delta     BIGINT,
    wal_bytes_delta             NUMERIC,
    hit_ratio_pct               NUMERIC,
    time_per_call_ms            DOUBLE PRECISION
)
LANGUAGE sql STABLE AS $$
    WITH
    start_snap AS (
        SELECT ss.*, s.captured_at
        FROM pgfr_record.statement_snapshots ss
        JOIN pgfr_record.snapshots s ON s.id = ss.snapshot_id
        WHERE s.captured_at <= p_start_time
        ORDER BY s.captured_at DESC
        LIMIT 1000
    ),
    end_snap AS (
        SELECT ss.*, s.captured_at
        FROM pgfr_record.statement_snapshots ss
        JOIN pgfr_record.snapshots s ON s.id = ss.snapshot_id
        WHERE s.captured_at >= p_end_time
        ORDER BY s.captured_at ASC
        LIMIT 1000
    ),
    matched AS (
        SELECT
            e.queryid,
            COALESCE(e.query_preview, s.query_preview) AS query_preview,
            s.calls AS calls_start,
            e.calls AS calls_end,
            s.total_exec_time AS total_exec_time_start,
            e.total_exec_time AS total_exec_time_end,
            s.mean_exec_time AS mean_exec_time_start,
            e.mean_exec_time AS mean_exec_time_end,
            s.rows AS rows_start,
            e.rows AS rows_end,
            s.shared_blks_hit AS shared_blks_hit_start,
            e.shared_blks_hit AS shared_blks_hit_end,
            s.shared_blks_read AS shared_blks_read_start,
            e.shared_blks_read AS shared_blks_read_end,
            s.shared_blks_written AS shared_blks_written_start,
            e.shared_blks_written AS shared_blks_written_end,
            s.temp_blks_read AS temp_blks_read_start,
            e.temp_blks_read AS temp_blks_read_end,
            s.temp_blks_written AS temp_blks_written_start,
            e.temp_blks_written AS temp_blks_written_end,
            s.wal_bytes AS wal_bytes_start,
            e.wal_bytes AS wal_bytes_end
        FROM end_snap e
        LEFT JOIN start_snap s ON s.queryid = e.queryid AND s.dbid = e.dbid
    )
    SELECT
        m.queryid,
        m.query_preview,
        COALESCE(m.calls_start, 0),
        m.calls_end,
        m.calls_end - COALESCE(m.calls_start, 0),
        COALESCE(m.total_exec_time_start, 0),
        m.total_exec_time_end,
        m.total_exec_time_end - COALESCE(m.total_exec_time_start, 0),
        m.mean_exec_time_start,
        m.mean_exec_time_end,
        m.rows_end - COALESCE(m.rows_start, 0),
        m.shared_blks_hit_end - COALESCE(m.shared_blks_hit_start, 0),
        m.shared_blks_read_end - COALESCE(m.shared_blks_read_start, 0),
        m.shared_blks_written_end - COALESCE(m.shared_blks_written_start, 0),
        m.temp_blks_read_end - COALESCE(m.temp_blks_read_start, 0),
        m.temp_blks_written_end - COALESCE(m.temp_blks_written_start, 0),
        m.wal_bytes_end - COALESCE(m.wal_bytes_start, 0),
        CASE
            WHEN (m.shared_blks_hit_end - COALESCE(m.shared_blks_hit_start, 0) +
                  m.shared_blks_read_end - COALESCE(m.shared_blks_read_start, 0)) > 0
            THEN round(
                100.0 * (m.shared_blks_hit_end - COALESCE(m.shared_blks_hit_start, 0)) /
                (m.shared_blks_hit_end - COALESCE(m.shared_blks_hit_start, 0) +
                 m.shared_blks_read_end - COALESCE(m.shared_blks_read_start, 0)), 1
            )
            ELSE NULL
        END,
        CASE
            WHEN (m.calls_end - COALESCE(m.calls_start, 0)) > 0
            THEN (m.total_exec_time_end - COALESCE(m.total_exec_time_start, 0)) /
                 (m.calls_end - COALESCE(m.calls_start, 0))
            ELSE NULL
        END
    FROM matched m
    WHERE (m.total_exec_time_end - COALESCE(m.total_exec_time_start, 0)) >= p_min_delta_ms
    ORDER BY (m.total_exec_time_end - COALESCE(m.total_exec_time_start, 0)) DESC
    LIMIT p_limit
$$;
COMMENT ON FUNCTION pgfr_analyze.statement_compare(TIMESTAMPTZ, TIMESTAMPTZ, DOUBLE PRECISION, INTEGER) IS 'Compares statement execution metrics between two snapshots, calculating performance deltas.';

-- Retrieves active session details at a specific point in time

-- ==========================================================
-- pgfr_analyze/sql/03_activity_storms_regressions.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

CREATE OR REPLACE FUNCTION pgfr_analyze.activity_at(p_timestamp TIMESTAMPTZ)
RETURNS TABLE(
    sample_captured_at      TIMESTAMPTZ,
    sample_offset_seconds   NUMERIC,
    active_sessions         INTEGER,
    waiting_sessions        INTEGER,
    idle_in_transaction     INTEGER,
    top_wait_event_1        TEXT,
    top_wait_count_1        INTEGER,
    top_wait_event_2        TEXT,
    top_wait_count_2        INTEGER,
    top_wait_event_3        TEXT,
    top_wait_count_3        INTEGER,
    blocked_pids            INTEGER,
    longest_blocked_duration INTERVAL,
    vacuums_running         INTEGER,
    copies_running          INTEGER,
    indexes_building        INTEGER,
    analyzes_running        INTEGER,
    snapshot_captured_at    TIMESTAMPTZ,
    snapshot_offset_seconds NUMERIC,
    autovacuum_workers      INTEGER,
    checkpoint_occurred     BOOLEAN
)
LANGUAGE sql STABLE AS $$
    WITH
    -- find nearest sample_ts from v2 ring (or fallback to legacy slot)
    nearest_ts AS (
        select captured_at, offset_secs, null::integer as slot_id, sample_ts
        from (
            -- v2: activity_samples
            select pgfr_record.epoch() + sample_ts * interval '1 second' as captured_at,
                   abs(extract(epoch from (
                       pgfr_record.epoch() + sample_ts * interval '1 second' - p_timestamp
                   ))) as offset_secs,
                   sample_ts
            from pgfr_record.activity_samples
            order by offset_secs
            limit 1
        ) v2
        union all
        select captured_at, offset_secs, slot_id, null::int4 as sample_ts
        from (
            -- legacy ring
            select slot_id, captured_at,
                   abs(extract(epoch from (captured_at - p_timestamp))) as offset_secs
            from pgfr_record.samples_ring
            order by offset_secs
            limit 1
        ) leg
        order by offset_secs
        limit 1
    ),
    nearest_sample AS (
        select captured_at,
               offset_secs,
               slot_id,
               extract(epoch from captured_at - pgfr_record.epoch())::int4 as sample_ts
        from nearest_ts
    ),
    nearest_snapshot AS (
        SELECT s.id, s.captured_at, s.autovacuum_workers,
               (s.checkpoint_time IS DISTINCT FROM prev.checkpoint_time) AS checkpoint_occurred,
               ABS(EXTRACT(EPOCH FROM (s.captured_at - p_timestamp))) AS offset_secs
        FROM pgfr_record.snapshots s
        LEFT JOIN pgfr_record.snapshots prev ON prev.id = (
            SELECT MAX(id) FROM pgfr_record.snapshots WHERE id < s.id
        )
        ORDER BY ABS(EXTRACT(EPOCH FROM (s.captured_at - p_timestamp)))
        LIMIT 1
    ),
    sample_waits AS (
        -- v2: decode wait_samples integer array
        SELECT wem.type || ':' || wem.event AS wait_event,
               abs(ws.data[i + 1]) AS count
        FROM pgfr_record.wait_samples ws
        cross join nearest_sample ns
        cross join generate_subscripts(ws.data, 1) AS i
        join pgfr_record.wait_event_map wem ON wem.id = abs(ws.data[i])::smallint
        WHERE ws.data[i] < 0
          AND ws.sample_ts = ns.sample_ts
          AND wem.state NOT IN ('idle', 'idle in transaction')
        UNION ALL
        -- legacy ring
        SELECT w.wait_event_type || ':' || w.wait_event AS wait_event, w.count
        FROM pgfr_record.wait_samples_ring w
        JOIN nearest_sample ns ON ns.slot_id = w.slot_id
        WHERE w.state NOT IN ('idle', 'idle in transaction')
    ),
    wait_ranked AS (
        select wait_event, sum(count) as count
        from sample_waits
        group by wait_event
        order by count desc
        limit 3
    ),
    wait_array AS (
        SELECT array_agg(wait_event ORDER BY count DESC) AS events,
               array_agg(count ORDER BY count DESC) AS counts
        FROM wait_ranked
    ),
    sample_activity AS (
        SELECT
            count(*) FILTER (WHERE state = 'active') AS active_sessions,
            count(*) FILTER (WHERE wait_event IS NOT NULL) AS waiting_sessions,
            count(*) FILTER (WHERE state = 'idle in transaction') AS idle_in_transaction
        FROM (
            -- v2
            SELECT a.state, a.wait_event
            FROM pgfr_record.activity_samples a
            JOIN nearest_sample ns ON a.sample_ts = ns.sample_ts
            UNION ALL
            -- legacy
            SELECT a.state, a.wait_event
            FROM pgfr_record.activity_samples_ring a
            JOIN nearest_sample ns ON ns.slot_id = a.slot_id
        ) combined
    ),
    sample_locks AS (
        SELECT
            count(DISTINCT blocked_pid) AS blocked_pids,
            max(blocked_duration) AS longest_blocked
        FROM (
            -- v2
            SELECT ls.blocked_pid,
                   ls.blocked_duration_s * interval '1 second' AS blocked_duration
            FROM pgfr_record.lock_samples ls
            JOIN nearest_sample ns ON ls.sample_ts = ns.sample_ts
            UNION ALL
            -- legacy
            SELECT l.blocked_pid, l.blocked_duration
            FROM pgfr_record.lock_samples_ring l
            JOIN nearest_sample ns ON ns.slot_id = l.slot_id
        ) combined
    ),
    sample_progress AS (
        SELECT 0 AS vacuums, 0 AS copies, 0 AS indexes, 0 AS analyzes
    )
    SELECT
        ns.captured_at,
        ns.offset_secs::numeric,
        COALESCE(sa.active_sessions, 0)::integer,
        COALESCE(sa.waiting_sessions, 0)::integer,
        COALESCE(sa.idle_in_transaction, 0)::integer,
        wa.events[1],
        wa.counts[1]::integer,
        wa.events[2],
        wa.counts[2]::integer,
        wa.events[3],
        wa.counts[3]::integer,
        COALESCE(sl.blocked_pids, 0)::integer,
        sl.longest_blocked,
        COALESCE(sp.vacuums, 0)::integer,
        COALESCE(sp.copies, 0)::integer,
        COALESCE(sp.indexes, 0)::integer,
        COALESCE(sp.analyzes, 0)::integer,
        sn.captured_at,
        sn.offset_secs::numeric,
        sn.autovacuum_workers,
        sn.checkpoint_occurred
    FROM nearest_sample ns
    CROSS JOIN nearest_snapshot sn
    CROSS JOIN sample_activity sa
    CROSS JOIN sample_locks sl
    CROSS JOIN sample_progress sp
    CROSS JOIN wait_array wa
$$;
COMMENT ON FUNCTION pgfr_analyze.activity_at(TIMESTAMPTZ) IS 'Retrieves active session details at a specific point in time.';

-- Detects query storms by comparing recent query execution counts to baseline
CREATE OR REPLACE FUNCTION pgfr_analyze.detect_query_storms(
    p_lookback INTERVAL DEFAULT NULL,
    p_threshold_multiplier NUMERIC DEFAULT NULL
)
RETURNS TABLE(
    queryid BIGINT,
    query_fingerprint TEXT,
    storm_type TEXT,
    severity TEXT,
    recent_count BIGINT,
    baseline_count BIGINT,
    multiplier NUMERIC
)
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_lookback INTERVAL;
    v_threshold NUMERIC;
    v_baseline_days INTEGER;
    v_low_max NUMERIC;
    v_medium_max NUMERIC;
    v_high_max NUMERIC;
BEGIN
    -- Get configuration with defaults
    v_lookback := COALESCE(
        p_lookback,
        pgfr_record._get_config('storm_lookback_interval', '1 hour')::interval
    );
    v_threshold := COALESCE(
        p_threshold_multiplier,
        pgfr_record._get_config('storm_threshold_multiplier', '3.0')::numeric
    );
    v_baseline_days := COALESCE(
        pgfr_record._get_config('storm_baseline_days', '7')::integer,
        7
    );

    -- Get severity thresholds
    v_low_max := pgfr_record._get_config('storm_severity_low_max', '5.0')::numeric;
    v_medium_max := pgfr_record._get_config('storm_severity_medium_max', '10.0')::numeric;
    v_high_max := pgfr_record._get_config('storm_severity_high_max', '50.0')::numeric;

    RETURN QUERY
    WITH recent_stats AS (
        -- Recent query call deltas from statement_snapshots
        SELECT
            ss.queryid,
            left(ss.query_preview, 100) AS query_preview,
            SUM(COALESCE(ss.calls_delta, 0)) AS total_calls
        FROM pgfr_record.statement_snapshots ss
        JOIN pgfr_record.snapshots s ON s.id = ss.snapshot_id
        WHERE s.captured_at >= now() - v_lookback
        GROUP BY ss.queryid, left(ss.query_preview, 100)
    ),
    baseline_stats AS (
        -- Baseline query call deltas (over baseline period, excluding recent)
        SELECT
            ss.queryid,
            AVG(COALESCE(ss.calls_delta, 0)) AS avg_calls,
            COUNT(DISTINCT date_trunc('day', s.captured_at)) AS days_sampled
        FROM pgfr_record.statement_snapshots ss
        JOIN pgfr_record.snapshots s ON s.id = ss.snapshot_id
        WHERE s.captured_at >= now() - (v_baseline_days || ' days')::interval
          AND s.captured_at < now() - v_lookback
        GROUP BY ss.queryid
        HAVING COUNT(DISTINCT date_trunc('day', s.captured_at)) >= 2  -- Need at least 2 days of baseline
    ),
    storms AS (
        SELECT
            r.queryid,
            r.query_preview AS query_fingerprint,
            CASE
                WHEN r.query_preview ILIKE '%RETRY%' OR r.query_preview ILIKE '%FOR UPDATE%'
                    THEN 'RETRY_STORM'
                WHEN r.total_calls > COALESCE(b.avg_calls, 0) * 10
                    THEN 'CACHE_MISS'
                WHEN r.total_calls > COALESCE(b.avg_calls, 1) * v_threshold
                    THEN 'SPIKE'
                ELSE 'NORMAL'
            END AS storm_type,
            r.total_calls::BIGINT AS recent_count,
            COALESCE(b.avg_calls, 0)::BIGINT AS baseline_count,
            CASE
                WHEN COALESCE(b.avg_calls, 0) > 0
                THEN ROUND(r.total_calls::numeric / b.avg_calls, 2)
                ELSE NULL
            END AS multiplier
        FROM recent_stats r
        LEFT JOIN baseline_stats b ON b.queryid = r.queryid
        WHERE r.total_calls > COALESCE(b.avg_calls, 1) * v_threshold
           OR (r.query_preview ILIKE '%RETRY%' OR r.query_preview ILIKE '%FOR UPDATE%')
    )
    SELECT
        st.queryid,
        st.query_fingerprint,
        st.storm_type,
        CASE
            WHEN st.storm_type = 'RETRY_STORM' THEN 'CRITICAL'
            WHEN st.multiplier > v_high_max THEN 'CRITICAL'
            WHEN st.multiplier > v_medium_max THEN 'HIGH'
            WHEN st.multiplier > v_low_max THEN 'MEDIUM'
            ELSE 'LOW'
        END AS severity,
        st.recent_count,
        st.baseline_count,
        st.multiplier
    FROM storms st
    ORDER BY
        CASE
            WHEN st.storm_type = 'RETRY_STORM' THEN 1
            WHEN st.multiplier > v_high_max THEN 2
            WHEN st.multiplier > v_medium_max THEN 3
            WHEN st.multiplier > v_low_max THEN 4
            ELSE 5
        END,
        st.recent_count DESC;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.detect_query_storms(INTERVAL, NUMERIC) IS 'Detect query storms by comparing recent execution counts to baseline. Classifies as RETRY_STORM, CACHE_MISS, SPIKE, or NORMAL with severity levels (LOW, MEDIUM, HIGH, CRITICAL).';

-- Diagnose probable causes for a query's performance regression
CREATE OR REPLACE FUNCTION pgfr_analyze._diagnose_regression_causes(
    p_queryid BIGINT
)
RETURNS TEXT[]
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_causes TEXT[] := ARRAY[]::TEXT[];
    v_stats RECORD;
    v_recent_snapshot RECORD;
BEGIN
    -- Get current stats from pg_stat_statements
    BEGIN
        SELECT
            temp_blks_written,
            shared_blks_hit,
            shared_blks_read,
            rows
        INTO v_stats
        FROM pg_stat_statements
        WHERE queryid = p_queryid
        LIMIT 1;
    EXCEPTION
        WHEN undefined_table THEN
            v_stats := NULL;
    END;

    IF v_stats IS NOT NULL THEN
        -- Check for temp file spills
        IF COALESCE(v_stats.temp_blks_written, 0) > 0 THEN
            v_causes := array_append(v_causes, 'Query is spilling to disk (temp files) - consider increasing work_mem');
        END IF;

        -- Check cache hit ratio
        IF v_stats.shared_blks_hit IS NOT NULL AND v_stats.shared_blks_read IS NOT NULL THEN
            IF v_stats.shared_blks_hit + v_stats.shared_blks_read > 0 THEN
                IF v_stats.shared_blks_hit::numeric / (v_stats.shared_blks_hit + v_stats.shared_blks_read) < 0.9 THEN
                    v_causes := array_append(v_causes, 'Low cache hit ratio - check shared_buffers or index usage');
                END IF;
            END IF;
        END IF;
    END IF;

    -- Check for recent checkpoint activity
    SELECT
        checkpoint_time,
        ckpt_write_time,
        ckpt_sync_time
    INTO v_recent_snapshot
    FROM pgfr_record.snapshots
    WHERE captured_at >= now() - interval '1 hour'
    ORDER BY captured_at DESC
    LIMIT 1;

    IF v_recent_snapshot IS NOT NULL THEN
        IF v_recent_snapshot.checkpoint_time >= now() - interval '5 minutes' THEN
            v_causes := array_append(v_causes, 'Recent checkpoint activity may be affecting I/O');
        END IF;
    END IF;

    -- Default causes if nothing specific found
    IF array_length(v_causes, 1) IS NULL THEN
        v_causes := array_append(v_causes, 'Statistics may be out of date - consider ANALYZE on involved tables');
        v_causes := array_append(v_causes, 'Query plan may have changed - check with EXPLAIN');
    END IF;

    RETURN v_causes;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze._diagnose_regression_causes(BIGINT) IS 'Internal: Analyze a query to suggest probable causes for performance regression.';

-- Detects performance regressions by comparing recent query metrics to baseline
CREATE OR REPLACE FUNCTION pgfr_analyze.detect_regressions(
    p_lookback INTERVAL DEFAULT NULL,
    p_threshold_pct NUMERIC DEFAULT NULL
)
RETURNS TABLE(
    queryid BIGINT,
    query_fingerprint TEXT,
    severity TEXT,
    baseline_avg_ms NUMERIC,
    current_avg_ms NUMERIC,
    change_pct NUMERIC,
    baseline_avg_buffers NUMERIC,
    current_avg_buffers NUMERIC,
    buffer_change_pct NUMERIC,
    detection_metric TEXT,
    probable_causes TEXT[]
)
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_lookback INTERVAL;
    v_threshold_pct NUMERIC;
    v_baseline_days INTEGER;
    v_low_max NUMERIC;
    v_medium_max NUMERIC;
    v_high_max NUMERIC;
    v_detection_metric TEXT;
BEGIN
    -- Get configuration with defaults
    v_lookback := COALESCE(
        p_lookback,
        pgfr_record._get_config('regression_lookback_interval', '1 hour')::interval
    );
    v_threshold_pct := COALESCE(
        p_threshold_pct,
        pgfr_record._get_config('regression_threshold_pct', '50.0')::numeric
    );
    v_baseline_days := COALESCE(
        pgfr_record._get_config('regression_baseline_days', '7')::integer,
        7
    );

    -- Get severity thresholds (percentage-based)
    v_low_max := pgfr_record._get_config('regression_severity_low_max', '200.0')::numeric;
    v_medium_max := pgfr_record._get_config('regression_severity_medium_max', '500.0')::numeric;
    v_high_max := pgfr_record._get_config('regression_severity_high_max', '1000.0')::numeric;

    -- Get detection metric (default to buffers)
    v_detection_metric := pgfr_record._get_config('regression_detection_metric', 'buffers');

    RETURN QUERY
    WITH recent_stats AS (
        -- Recent query metrics from statement_snapshots
        SELECT
            ss.queryid,
            left(ss.query_preview, 100) AS query_preview,
            AVG(ss.mean_exec_time) AS avg_mean_time,
            STDDEV(ss.mean_exec_time) AS stddev_mean_time,
            AVG(COALESCE(ss.shared_blks_hit_delta, 0) + COALESCE(ss.shared_blks_read_delta, 0)
                + COALESCE(ss.temp_blks_read_delta, 0)) AS avg_total_buffers,
            STDDEV(COALESCE(ss.shared_blks_hit_delta, 0) + COALESCE(ss.shared_blks_read_delta, 0)
                + COALESCE(ss.temp_blks_read_delta, 0)) AS stddev_total_buffers,
            COUNT(*) AS sample_count
        FROM pgfr_record.statement_snapshots ss
        JOIN pgfr_record.snapshots s ON s.id = ss.snapshot_id
        WHERE s.captured_at >= now() - v_lookback
          AND ss.mean_exec_time IS NOT NULL
          AND ss.mean_exec_time > 0
        GROUP BY ss.queryid, left(ss.query_preview, 100)
    ),
    baseline_stats AS (
        -- Baseline query metrics (over baseline period, excluding recent)
        SELECT
            ss.queryid,
            AVG(ss.mean_exec_time) AS avg_mean_time,
            STDDEV(ss.mean_exec_time) AS stddev_mean_time,
            AVG(COALESCE(ss.shared_blks_hit_delta, 0) + COALESCE(ss.shared_blks_read_delta, 0)
                + COALESCE(ss.temp_blks_read_delta, 0)) AS avg_total_buffers,
            STDDEV(COALESCE(ss.shared_blks_hit_delta, 0) + COALESCE(ss.shared_blks_read_delta, 0)
                + COALESCE(ss.temp_blks_read_delta, 0)) AS stddev_total_buffers,
            COUNT(DISTINCT date_trunc('day', s.captured_at)) AS days_sampled
        FROM pgfr_record.statement_snapshots ss
        JOIN pgfr_record.snapshots s ON s.id = ss.snapshot_id
        WHERE s.captured_at >= now() - (v_baseline_days || ' days')::interval
          AND s.captured_at < now() - v_lookback
          AND ss.mean_exec_time IS NOT NULL
          AND ss.mean_exec_time > 0
        GROUP BY ss.queryid
        HAVING COUNT(DISTINCT date_trunc('day', s.captured_at)) >= 2  -- Need at least 2 days of baseline
    ),
    regressions AS (
        SELECT
            r.queryid,
            r.query_preview AS query_fingerprint,
            -- Timing metrics
            b.avg_mean_time::numeric AS baseline_avg_ms,
            r.avg_mean_time::numeric AS current_avg_ms,
            ROUND(((r.avg_mean_time - b.avg_mean_time) / NULLIF(b.avg_mean_time, 0))::numeric * 100, 2) AS time_change_pct,
            -- Buffer metrics
            b.avg_total_buffers::numeric AS baseline_avg_buffers,
            r.avg_total_buffers::numeric AS current_avg_buffers,
            ROUND(((r.avg_total_buffers - b.avg_total_buffers) / NULLIF(b.avg_total_buffers, 0))::numeric * 100, 2) AS buffer_change_pct,
            -- Z-score calculation for statistical significance (based on detection metric)
            CASE
                WHEN v_detection_metric = 'time' THEN
                    CASE
                        WHEN COALESCE(b.stddev_mean_time, 0) > 0
                        THEN (r.avg_mean_time - b.avg_mean_time) / b.stddev_mean_time
                        ELSE 0
                    END
                ELSE
                    CASE
                        WHEN COALESCE(b.stddev_total_buffers, 0) > 0
                        THEN (r.avg_total_buffers - b.avg_total_buffers) / b.stddev_total_buffers
                        ELSE 0
                    END
            END AS z_score,
            -- Change percentage based on detection metric
            CASE
                WHEN v_detection_metric = 'time' THEN
                    ROUND(((r.avg_mean_time - b.avg_mean_time) / NULLIF(b.avg_mean_time, 0))::numeric * 100, 2)
                ELSE
                    ROUND(((r.avg_total_buffers - b.avg_total_buffers) / NULLIF(b.avg_total_buffers, 0))::numeric * 100, 2)
            END AS primary_change_pct
        FROM recent_stats r
        JOIN baseline_stats b ON b.queryid = r.queryid
        WHERE r.sample_count >= 2  -- Need multiple samples to be confident
          AND CASE
              WHEN v_detection_metric = 'time' THEN
                  r.avg_mean_time > b.avg_mean_time * (1 + v_threshold_pct / 100)
              ELSE
                  COALESCE(r.avg_total_buffers, 0) > COALESCE(b.avg_total_buffers, 0) * (1 + v_threshold_pct / 100)
                  AND b.avg_total_buffers IS NOT NULL AND b.avg_total_buffers > 0
          END
    )
    SELECT
        reg.queryid,
        reg.query_fingerprint,
        CASE
            WHEN reg.primary_change_pct > v_high_max THEN 'CRITICAL'
            WHEN reg.primary_change_pct > v_medium_max THEN 'HIGH'
            WHEN reg.primary_change_pct > v_low_max THEN 'MEDIUM'
            ELSE 'LOW'
        END AS severity,
        ROUND(reg.baseline_avg_ms, 2) AS baseline_avg_ms,
        ROUND(reg.current_avg_ms, 2) AS current_avg_ms,
        reg.time_change_pct AS change_pct,
        ROUND(reg.baseline_avg_buffers, 0) AS baseline_avg_buffers,
        ROUND(reg.current_avg_buffers, 0) AS current_avg_buffers,
        reg.buffer_change_pct,
        v_detection_metric AS detection_metric,
        pgfr_analyze._diagnose_regression_causes(reg.queryid) AS probable_causes
    FROM regressions reg
    WHERE reg.z_score > 2 OR reg.primary_change_pct > v_medium_max  -- Statistical filter or significant change
    ORDER BY
        CASE
            WHEN reg.primary_change_pct > v_high_max THEN 1
            WHEN reg.primary_change_pct > v_medium_max THEN 2
            WHEN reg.primary_change_pct > v_low_max THEN 3
            ELSE 4
        END,
        reg.primary_change_pct DESC;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.detect_regressions(INTERVAL, NUMERIC) IS 'Detect performance regressions using buffer metrics (default) or timing. Classifies severity based on percentage change (LOW <200%, MEDIUM <500%, HIGH <1000%, CRITICAL >1000%). Configure via regression_detection_metric.';
-- Generates a health report of flight recorder operations, including collection performance metrics,
-- success rates, and schema size with qualitative assessments

-- ==========================================================
-- pgfr_analyze/sql/04_reports.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

CREATE OR REPLACE FUNCTION pgfr_analyze.performance_report(p_lookback_interval INTERVAL DEFAULT '24 hours')
RETURNS TABLE(
    metric TEXT,
    value TEXT,
    assessment TEXT
)
LANGUAGE plpgsql AS $$
DECLARE
    v_avg_sample_ms NUMERIC;
    v_max_sample_ms INTEGER;
    v_avg_snapshot_ms NUMERIC;
    v_max_snapshot_ms INTEGER;
    v_total_collections INTEGER;
    v_failed_collections INTEGER;
    v_skipped_collections INTEGER;
    v_schema_size_mb NUMERIC;
BEGIN
    SELECT
        avg(duration_ms) FILTER (WHERE collection_type = 'sample' AND success = true AND skipped = false),
        max(duration_ms) FILTER (WHERE collection_type = 'sample' AND success = true AND skipped = false),
        avg(duration_ms) FILTER (WHERE collection_type = 'snapshot' AND success = true AND skipped = false),
        max(duration_ms) FILTER (WHERE collection_type = 'snapshot' AND success = true AND skipped = false),
        count(*),
        count(*) FILTER (WHERE success = false),
        count(*) FILTER (WHERE skipped = true)
    INTO v_avg_sample_ms, v_max_sample_ms, v_avg_snapshot_ms, v_max_snapshot_ms,
         v_total_collections, v_failed_collections, v_skipped_collections
    FROM pgfr_record.collection_stats
    WHERE started_at > now() - p_lookback_interval;
    SELECT schema_size_mb INTO v_schema_size_mb FROM pgfr_record._check_schema_size();
    RETURN QUERY SELECT
        'Avg Sample Duration'::text,
        COALESCE(round(v_avg_sample_ms)::text || ' ms', 'N/A'),
        CASE
            WHEN v_avg_sample_ms IS NULL THEN 'No data'
            WHEN v_avg_sample_ms < 100 THEN 'Excellent'
            WHEN v_avg_sample_ms < 500 THEN 'Good'
            WHEN v_avg_sample_ms < 1000 THEN 'Acceptable'
            ELSE 'Poor - consider emergency mode'
        END::text;
    RETURN QUERY SELECT
        'Max Sample Duration'::text,
        COALESCE(v_max_sample_ms::text || ' ms', 'N/A'),
        CASE
            WHEN v_max_sample_ms IS NULL THEN 'No data'
            WHEN v_max_sample_ms < 1000 THEN 'Good'
            WHEN v_max_sample_ms < 5000 THEN 'Acceptable'
            ELSE 'Circuit breaker may trip'
        END::text;
    RETURN QUERY SELECT
        'Avg Snapshot Duration'::text,
        COALESCE(round(v_avg_snapshot_ms)::text || ' ms', 'N/A'),
        CASE
            WHEN v_avg_snapshot_ms IS NULL THEN 'No data'
            WHEN v_avg_snapshot_ms < 500 THEN 'Excellent'
            WHEN v_avg_snapshot_ms < 2000 THEN 'Good'
            WHEN v_avg_snapshot_ms < 5000 THEN 'Acceptable'
            ELSE 'Poor'
        END::text;
    RETURN QUERY SELECT
        'Schema Size'::text,
        round(v_schema_size_mb)::text || ' MB',
        CASE
            WHEN v_schema_size_mb < 1000 THEN 'Healthy'
            WHEN v_schema_size_mb < 5000 THEN 'Good'
            WHEN v_schema_size_mb < 8000 THEN 'Consider cleanup()'
            ELSE 'Run cleanup() soon'
        END::text;
    RETURN QUERY SELECT
        'Collection Success Rate'::text,
        format('%s%% (%s/%s)',
               round(((v_total_collections - v_failed_collections)::numeric / NULLIF(v_total_collections, 0)) * 100, 1)::text,
               v_total_collections - v_failed_collections,
               v_total_collections),
        CASE
            WHEN v_total_collections = 0 THEN 'No collections'
            WHEN v_failed_collections = 0 THEN 'Perfect'
            WHEN (v_failed_collections::numeric / v_total_collections) < 0.01 THEN 'Excellent'
            WHEN (v_failed_collections::numeric / v_total_collections) < 0.05 THEN 'Good'
            ELSE 'Issues detected'
        END::text;
    RETURN QUERY SELECT
        'Skipped Collections'::text,
        v_skipped_collections::text,
        CASE
            WHEN v_skipped_collections = 0 THEN 'No skips'
            WHEN v_skipped_collections < 5 THEN 'Minimal'
            WHEN v_skipped_collections < 20 THEN 'Moderate - check system load'
            ELSE 'Significant - system under stress'
        END::text;
    RETURN QUERY SELECT
        'Section Success Rate'::text,
        COALESCE(
            round(100.0 * avg(sections_succeeded::numeric / NULLIF(sections_total, 0)), 1)::text || '%',
            'N/A'
        ),
        CASE
            WHEN avg(sections_succeeded::numeric / NULLIF(sections_total, 0)) IS NULL THEN 'No data'
            WHEN avg(sections_succeeded::numeric / NULLIF(sections_total, 0)) >= 0.95 THEN 'Excellent'
            WHEN avg(sections_succeeded::numeric / NULLIF(sections_total, 0)) >= 0.90 THEN 'Good'
            WHEN avg(sections_succeeded::numeric / NULLIF(sections_total, 0)) >= 0.75 THEN 'Fair - some section failures'
            ELSE 'Poor - frequent section failures'
        END::text
    FROM pgfr_record.collection_stats
    WHERE started_at > now() - p_lookback_interval
      AND sections_total IS NOT NULL;
    RETURN QUERY
    WITH recent AS (
        SELECT duration_ms
        FROM pgfr_record.collection_stats
        WHERE collection_type = 'sample'
          AND success = true
          AND skipped = false
          AND started_at > now() - p_lookback_interval
        ORDER BY started_at DESC
        LIMIT 10
    ),
    baseline AS (
        SELECT duration_ms
        FROM pgfr_record.collection_stats
        WHERE collection_type = 'sample'
          AND success = true
          AND skipped = false
          AND started_at > now() - p_lookback_interval
        ORDER BY started_at DESC
        LIMIT 100 OFFSET 50
    )
    SELECT
        'Performance Trend'::text,
        COALESCE(
            CASE
                WHEN (SELECT avg(duration_ms) FROM recent) IS NULL OR (SELECT avg(duration_ms) FROM baseline) IS NULL THEN 'Insufficient data'
                ELSE format('%s → %s ms (%s%s)',
                    round((SELECT avg(duration_ms) FROM baseline))::text,
                    round((SELECT avg(duration_ms) FROM recent))::text,
                    CASE WHEN ((SELECT avg(duration_ms) FROM recent) - (SELECT avg(duration_ms) FROM baseline)) > 0 THEN '+' ELSE '' END,
                    round((((SELECT avg(duration_ms) FROM recent) - (SELECT avg(duration_ms) FROM baseline)) / NULLIF((SELECT avg(duration_ms) FROM baseline), 0)) * 100, 1)::text || '%'
                )
            END,
            'N/A'
        ),
        CASE
            WHEN (SELECT avg(duration_ms) FROM recent) IS NULL OR (SELECT avg(duration_ms) FROM baseline) IS NULL THEN 'Need more data'
            WHEN (SELECT avg(duration_ms) FROM recent) > (SELECT avg(duration_ms) FROM baseline) * 1.5 THEN 'DEGRADING - investigate system load'
            WHEN (SELECT avg(duration_ms) FROM recent) < (SELECT avg(duration_ms) FROM baseline) * 0.7 THEN 'IMPROVING'
            ELSE 'STABLE'
        END::text;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.performance_report(INTERVAL) IS
'Reports on flight recorder collection performance: sample/snapshot timing, success rates, schema size, and qualitative health assessments over a configurable lookback window.';

-- Monitors flight recorder system health by checking for circuit breaker trips, schema size limits, collection failures, and stale data
-- Returns alerts with severity levels (CRITICAL/WARNING) and recommendations when thresholds are exceeded
CREATE OR REPLACE FUNCTION pgfr_analyze.check_alerts(p_lookback_interval INTERVAL DEFAULT '1 hour')
RETURNS TABLE(
    alert_type TEXT,
    severity TEXT,
    message TEXT,
    triggered_at TIMESTAMPTZ,
    recommendation TEXT
)
LANGUAGE plpgsql AS $$
DECLARE
    v_enabled BOOLEAN;
    v_cb_threshold INTEGER;
    v_cb_count INTEGER;
    v_schema_threshold_mb INTEGER;
    v_schema_size_mb NUMERIC;
BEGIN
    v_enabled := COALESCE(
        pgfr_record._get_config('alert_enabled', 'false')::boolean,
        false
    );
    IF NOT v_enabled THEN
        RETURN;
    END IF;
    v_cb_threshold := COALESCE(
        pgfr_record._get_config('alert_circuit_breaker_count', '5')::integer,
        5
    );
    SELECT count(*) INTO v_cb_count
    FROM pgfr_record.collection_stats
    WHERE skipped = true
      AND started_at > now() - p_lookback_interval
      AND skipped_reason LIKE '%Circuit breaker%';
    IF v_cb_count >= v_cb_threshold THEN
        RETURN QUERY SELECT
            'CIRCUIT_BREAKER_TRIPS'::text,
            'CRITICAL'::text,
            format('Circuit breaker tripped %s times in last %s', v_cb_count, p_lookback_interval),
            now(),
            'System under severe stress. Consider switching to emergency mode or disabling flight recorder temporarily.'::text;
    END IF;
    v_schema_threshold_mb := COALESCE(
        pgfr_record._get_config('alert_schema_size_mb', '8000')::integer,
        8000
    );
    SELECT schema_size_mb INTO v_schema_size_mb FROM pgfr_record._check_schema_size();
    IF v_schema_size_mb >= v_schema_threshold_mb THEN
        RETURN QUERY SELECT
            'SCHEMA_SIZE_HIGH'::text,
            'WARNING'::text,
            format('Schema size is %s MB (threshold: %s MB)', round(v_schema_size_mb)::text, v_schema_threshold_mb),
            now(),
            'Run pgfr_record.cleanup() to reclaim space.'::text;
    END IF;
    DECLARE
        v_recent_failures INTEGER;
    BEGIN
        SELECT count(*) INTO v_recent_failures
        FROM pgfr_record.collection_stats
        WHERE success = false
          AND started_at > now() - p_lookback_interval;
        IF v_recent_failures >= 5 THEN
            RETURN QUERY SELECT
                'COLLECTION_FAILURES'::text,
                'WARNING'::text,
                format('%s collection failures in last %s', v_recent_failures, p_lookback_interval),
                now(),
                'Check PostgreSQL logs for error details.'::text;
        END IF;
    END;
    DECLARE
        v_last_sample TIMESTAMPTZ;
    BEGIN
        SELECT max(captured_at) INTO v_last_sample FROM pgfr_record.samples_ring;
        IF v_last_sample IS NULL OR v_last_sample < now() - interval '15 minutes' THEN
            RETURN QUERY SELECT
                'STALE_DATA'::text,
                'CRITICAL'::text,
                CASE
                    WHEN v_last_sample IS NULL THEN 'No samples collected yet'
                    ELSE format('Last sample was %s ago', age(now(), v_last_sample))
                END,
                now(),
                'Check pg_cron jobs: SELECT * FROM cron.job WHERE jobname LIKE ''pgfr_%'''::text;
        END IF;
    END;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.check_alerts(INTERVAL) IS
'Monitors flight recorder system health and returns alerts with severity levels (CRITICAL/WARNING) for circuit breaker trips, schema size limits, collection failures, and stale data.';

-- Exports flight recorder diagnostic data as human-readable Markdown
-- Produces a report with tables that is legible to both humans and AI
CREATE OR REPLACE FUNCTION pgfr_analyze.report(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TEXT
LANGUAGE plpgsql AS $$
DECLARE
    v_result TEXT := '';
    v_version TEXT;
    v_row RECORD;
    v_count INTEGER;
BEGIN
    -- Get schema version from config
    SELECT value INTO v_version FROM pgfr_record.config WHERE key = 'schema_version';
    v_version := COALESCE(v_version, 'unknown');

    -- Header
    v_result := v_result || '# PostgreSQL Flight Recorder Report' || E'\n\n';
    v_result := v_result || '**Generated:** ' || to_char(now(), 'YYYY-MM-DD HH24:MI:SS TZ') || E'\n';
    v_result := v_result || '**Version:** ' || v_version || E'\n';
    v_result := v_result || '**Range:** ' || to_char(p_start_time, 'YYYY-MM-DD HH24:MI:SS') ||
                           ' to ' || to_char(p_end_time, 'YYYY-MM-DD HH24:MI:SS') || E'\n\n';
    v_result := v_result || 'Analyze this data. The database may be healthy—only flag genuine issues.' || E'\n\n';

    -- ==========================================================================
    -- Anomalies Section
    -- ==========================================================================
    v_result := v_result || '## Anomalies' || E'\n\n';

    SELECT count(*) INTO v_count FROM pgfr_analyze.anomaly_report(p_start_time, p_end_time);
    IF v_count = 0 THEN
        v_result := v_result || '**No anomalies detected.** System appears healthy.' || E'\n\n';
    ELSE
        v_result := v_result || '| Type | Severity | Description | Metric | Recommendation |' || E'\n';
        v_result := v_result || '|------|----------|-------------|--------|----------------|' || E'\n';
        FOR v_row IN SELECT * FROM pgfr_analyze.anomaly_report(p_start_time, p_end_time) LOOP
            v_result := v_result || '| ' ||
                COALESCE(v_row.anomaly_type, '-') || ' | ' ||
                COALESCE(v_row.severity, '-') || ' | ' ||
                COALESCE(v_row.description, '-') || ' | ' ||
                COALESCE(v_row.metric_value, '-') || ' | ' ||
                COALESCE(v_row.recommendation, '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Wait Event Summary Section
    -- ==========================================================================
    v_result := v_result || '## Wait Event Summary' || E'\n\n';

    SELECT count(*) INTO v_count FROM pgfr_analyze.wait_summary(p_start_time, p_end_time);
    IF v_count = 0 THEN
        v_result := v_result || '(no wait events recorded)' || E'\n\n';
    ELSE
        v_result := v_result || '| Backend | Event Type | Event | Samples | Avg Waiters | Max | % |' || E'\n';
        v_result := v_result || '|---------|------------|-------|---------|-------------|-----|---|' || E'\n';
        FOR v_row IN SELECT * FROM pgfr_analyze.wait_summary(p_start_time, p_end_time) LOOP
            v_result := v_result || '| ' ||
                COALESCE(v_row.backend_state, '-') || ' | ' ||
                COALESCE(v_row.wait_event_type, '-') || ' | ' ||
                COALESCE(v_row.wait_event, '-') || ' | ' ||
                COALESCE(v_row.sample_count::TEXT, '-') || ' | ' ||
                COALESCE(v_row.avg_waiters::TEXT, '-') || ' | ' ||
                COALESCE(v_row.max_waiters::TEXT, '-') || ' | ' ||
                COALESCE(v_row.pct_of_samples::TEXT, '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Snapshots Section
    -- ==========================================================================
    v_result := v_result || '## Snapshots' || E'\n\n';

    SELECT count(*) INTO v_count
    FROM pgfr_record.snapshots
    WHERE captured_at BETWEEN p_start_time AND p_end_time;

    IF v_count = 0 THEN
        v_result := v_result || '(no snapshots in range)' || E'\n\n';
    ELSE
        v_result := v_result || '| Captured At | WAL Bytes | Ckpt (Timed) | Ckpt (Req) | Backend Writes |' || E'\n';
        v_result := v_result || '|-------------|-----------|--------------|------------|----------------|' || E'\n';
        FOR v_row IN
            SELECT captured_at, wal_bytes, ckpt_timed, ckpt_requested, bgw_buffers_backend
            FROM pgfr_record.snapshots
            WHERE captured_at BETWEEN p_start_time AND p_end_time
            ORDER BY captured_at
        LOOP
            v_result := v_result || '| ' ||
                to_char(v_row.captured_at, 'YYYY-MM-DD HH24:MI:SS') || ' | ' ||
                COALESCE(pgfr_record._pretty_bytes(v_row.wal_bytes), '-') || ' | ' ||
                COALESCE(v_row.ckpt_timed::TEXT, '-') || ' | ' ||
                COALESCE(v_row.ckpt_requested::TEXT, '-') || ' | ' ||
                COALESCE(v_row.bgw_buffers_backend::TEXT, '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Table Hotspots Section
    -- ==========================================================================
    v_result := v_result || '## Table Hotspots' || E'\n\n';

    SELECT count(*) INTO v_count FROM pgfr_analyze.table_hotspots(p_start_time, p_end_time);
    IF v_count = 0 THEN
        v_result := v_result || '(no issues detected)' || E'\n\n';
    ELSE
        v_result := v_result || '| Schema | Table | Issue | Severity | Description | Recommendation |' || E'\n';
        v_result := v_result || '|--------|-------|-------|----------|-------------|----------------|' || E'\n';
        FOR v_row IN SELECT * FROM pgfr_analyze.table_hotspots(p_start_time, p_end_time) LOOP
            v_result := v_result || '| ' ||
                COALESCE(v_row.schemaname, '-') || ' | ' ||
                COALESCE(v_row.relname, '-') || ' | ' ||
                COALESCE(v_row.issue_type, '-') || ' | ' ||
                COALESCE(v_row.severity, '-') || ' | ' ||
                COALESCE(v_row.description, '-') || ' | ' ||
                COALESCE(v_row.recommendation, '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Index Efficiency Section
    -- ==========================================================================
    v_result := v_result || '## Index Efficiency' || E'\n\n';

    SELECT count(*) INTO v_count FROM pgfr_analyze.index_efficiency(p_start_time, p_end_time);
    IF v_count = 0 THEN
        v_result := v_result || '(no index activity in range)' || E'\n\n';
    ELSE
        v_result := v_result || '| Schema | Table | Index | Scans | Selectivity | Size | Scans/GB |' || E'\n';
        v_result := v_result || '|--------|-------|-------|-------|-------------|------|----------|' || E'\n';
        FOR v_row IN SELECT * FROM pgfr_analyze.index_efficiency(p_start_time, p_end_time) LOOP
            v_result := v_result || '| ' ||
                COALESCE(v_row.schemaname, '-') || ' | ' ||
                COALESCE(v_row.relname, '-') || ' | ' ||
                COALESCE(v_row.indexrelname, '-') || ' | ' ||
                COALESCE(v_row.idx_scan_delta::TEXT, '-') || ' | ' ||
                COALESCE(v_row.selectivity::TEXT || '%', '-') || ' | ' ||
                COALESCE(v_row.index_size, '-') || ' | ' ||
                COALESCE(v_row.scans_per_gb::TEXT, '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Statement Performance Section (requires pg_stat_statements)
    -- ==========================================================================
    v_result := v_result || '## Statement Performance' || E'\n\n';

    BEGIN
        SELECT count(*) INTO v_count FROM pgfr_analyze.statement_compare(p_start_time, p_end_time, 100, 25);
        IF v_count = 0 THEN
            v_result := v_result || '(no significant query changes)' || E'\n\n';
        ELSE
            v_result := v_result || '| Query | Calls Δ | Total Time Δ (ms) | Mean (ms) | Temp Writes | Hit % |' || E'\n';
            v_result := v_result || '|-------|---------|-------------------|-----------|-------------|-------|' || E'\n';
            FOR v_row IN
                SELECT * FROM pgfr_analyze.statement_compare(p_start_time, p_end_time, 100, 25)
                ORDER BY total_exec_time_delta_ms DESC NULLS LAST
            LOOP
                v_result := v_result || '| ' ||
                    COALESCE(left(v_row.query_preview, 60), '-') || ' | ' ||
                    COALESCE(v_row.calls_delta::TEXT, '-') || ' | ' ||
                    COALESCE(round(v_row.total_exec_time_delta_ms::NUMERIC, 1)::TEXT, '-') || ' | ' ||
                    COALESCE(round(v_row.mean_exec_time_end_ms::NUMERIC, 2)::TEXT, '-') || ' | ' ||
                    COALESCE(v_row.temp_blks_written_delta::TEXT, '-') || ' | ' ||
                    COALESCE(v_row.hit_ratio_pct::TEXT, '-') || ' |' || E'\n';
            END LOOP;
            v_result := v_result || E'\n';
        END IF;
    EXCEPTION
        WHEN undefined_table OR undefined_function THEN
            v_result := v_result || '(pg_stat_statements not available)' || E'\n\n';
    END;

    -- ==========================================================================
    -- Lock Contention Section
    -- ==========================================================================
    v_result := v_result || '## Lock Contention' || E'\n\n';

    SELECT count(*) INTO v_count
    FROM pgfr_record.lock_samples_archive
    WHERE captured_at BETWEEN p_start_time AND p_end_time;

    IF v_count = 0 THEN
        v_result := v_result || '(no lock contention recorded)' || E'\n\n';
    ELSE
        v_result := v_result || '| Time | Blocked PID | Blocking PID | Lock Type | Duration | Blocked Query |' || E'\n';
        v_result := v_result || '|------|-------------|--------------|-----------|----------|---------------|' || E'\n';
        FOR v_row IN
            SELECT *
            FROM pgfr_record.lock_samples_archive
            WHERE captured_at BETWEEN p_start_time AND p_end_time
            ORDER BY captured_at DESC
            LIMIT 50
        LOOP
            v_result := v_result || '| ' ||
                to_char(v_row.captured_at, 'HH24:MI:SS') || ' | ' ||
                COALESCE(v_row.blocked_pid::TEXT, '-') || ' | ' ||
                COALESCE(v_row.blocking_pid::TEXT, '-') || ' | ' ||
                COALESCE(v_row.lock_type, '-') || ' | ' ||
                COALESCE(v_row.blocked_duration::TEXT, '-') || ' | ' ||
                COALESCE(left(v_row.blocked_query_preview, 40), '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Long-Running Transactions Section
    -- ==========================================================================
    v_result := v_result || '## Long-Running Transactions' || E'\n\n';

    SELECT count(*) INTO v_count
    FROM pgfr_record.activity_samples_archive
    WHERE captured_at BETWEEN p_start_time AND p_end_time
      AND xact_start IS NOT NULL
      AND captured_at - xact_start > interval '5 minutes';

    IF v_count = 0 THEN
        v_result := v_result || '(no long-running transactions detected)' || E'\n\n';
    ELSE
        v_result := v_result || '| Time | PID | User | App | Transaction Age | State | Query Preview |' || E'\n';
        v_result := v_result || '|------|-----|------|-----|-----------------|-------|---------------|' || E'\n';
        FOR v_row IN
            SELECT DISTINCT ON (pid, xact_start)
                captured_at,
                pid,
                usename,
                application_name,
                captured_at - xact_start AS xact_age,
                state,
                query_preview
            FROM pgfr_record.activity_samples_archive
            WHERE captured_at BETWEEN p_start_time AND p_end_time
              AND xact_start IS NOT NULL
              AND captured_at - xact_start > interval '5 minutes'
            ORDER BY pid, xact_start, captured_at DESC
            LIMIT 25
        LOOP
            v_result := v_result || '| ' ||
                to_char(v_row.captured_at, 'HH24:MI:SS') || ' | ' ||
                COALESCE(v_row.pid::TEXT, '-') || ' | ' ||
                COALESCE(v_row.usename, '-') || ' | ' ||
                COALESCE(left(v_row.application_name, 15), '-') || ' | ' ||
                COALESCE(v_row.xact_age::TEXT, '-') || ' | ' ||
                COALESCE(v_row.state, '-') || ' | ' ||
                COALESCE(left(v_row.query_preview, 30), '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Vacuum Progress Section
    -- ==========================================================================
    v_result := v_result || '## Vacuum Progress' || E'\n\n';

    SELECT count(*) INTO v_count
    FROM pgfr_record.vacuum_progress_snapshots v
    JOIN pgfr_record.snapshots s ON s.id = v.snapshot_id
    WHERE s.captured_at BETWEEN p_start_time AND p_end_time;

    IF v_count = 0 THEN
        v_result := v_result || '(no vacuums captured during this period)' || E'\n\n';
    ELSE
        v_result := v_result || '| Time | Database | Table | Phase | % Scanned | % Vacuumed | Dead Tuples |' || E'\n';
        v_result := v_result || '|------|----------|-------|-------|-----------|------------|-------------|' || E'\n';
        FOR v_row IN
            SELECT
                s.captured_at,
                v.datname,
                v.relname,
                v.phase,
                CASE WHEN v.heap_blks_total > 0
                    THEN round(100.0 * v.heap_blks_scanned / v.heap_blks_total, 1)
                    ELSE NULL
                END AS pct_scanned,
                CASE WHEN v.heap_blks_total > 0
                    THEN round(100.0 * v.heap_blks_vacuumed / v.heap_blks_total, 1)
                    ELSE NULL
                END AS pct_vacuumed,
                v.num_dead_tuples
            FROM pgfr_record.vacuum_progress_snapshots v
            JOIN pgfr_record.snapshots s ON s.id = v.snapshot_id
            WHERE s.captured_at BETWEEN p_start_time AND p_end_time
            ORDER BY s.captured_at DESC
            LIMIT 25
        LOOP
            v_result := v_result || '| ' ||
                to_char(v_row.captured_at, 'HH24:MI:SS') || ' | ' ||
                COALESCE(v_row.datname, '-') || ' | ' ||
                COALESCE(v_row.relname, '-') || ' | ' ||
                COALESCE(v_row.phase, '-') || ' | ' ||
                COALESCE(v_row.pct_scanned::TEXT || '%', '-') || ' | ' ||
                COALESCE(v_row.pct_vacuumed::TEXT || '%', '-') || ' | ' ||
                COALESCE(v_row.num_dead_tuples::TEXT, '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Archiver Status Section
    -- ==========================================================================
    v_result := v_result || '## WAL Archiver Status' || E'\n\n';

    SELECT count(*) INTO v_count
    FROM pgfr_record.snapshots
    WHERE captured_at BETWEEN p_start_time AND p_end_time
      AND archived_count IS NOT NULL;

    IF v_count = 0 THEN
        v_result := v_result || '(archiving not enabled or no data in range)' || E'\n\n';
    ELSE
        -- Show summary: total archived, total failed, any failures
        FOR v_row IN
            SELECT
                min(archived_count) AS start_archived,
                max(archived_count) AS end_archived,
                max(archived_count) - min(archived_count) AS archived_delta,
                min(failed_count) AS start_failed,
                max(failed_count) AS end_failed,
                max(failed_count) - min(failed_count) AS failed_delta,
                max(last_failed_wal) AS last_failed_wal,
                max(last_failed_time) AS last_failed_time
            FROM pgfr_record.snapshots
            WHERE captured_at BETWEEN p_start_time AND p_end_time
              AND archived_count IS NOT NULL
        LOOP
            v_result := v_result || '| Metric | Value |' || E'\n';
            v_result := v_result || '|--------|-------|' || E'\n';
            v_result := v_result || '| WAL Files Archived | ' ||
                COALESCE(v_row.archived_delta::TEXT, '0') || ' |' || E'\n';
            v_result := v_result || '| Archive Failures | ' ||
                COALESCE(v_row.failed_delta::TEXT, '0') || ' |' || E'\n';
            IF v_row.failed_delta > 0 AND v_row.last_failed_wal IS NOT NULL THEN
                v_result := v_result || '| Last Failed WAL | ' ||
                    v_row.last_failed_wal || ' |' || E'\n';
                v_result := v_result || '| Last Failure Time | ' ||
                    to_char(v_row.last_failed_time, 'YYYY-MM-DD HH24:MI:SS') || ' |' || E'\n';
            END IF;
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Configuration Changes Section
    -- ==========================================================================
    v_result := v_result || '## Configuration Changes' || E'\n\n';

    SELECT count(*) INTO v_count FROM pgfr_analyze.config_changes(p_start_time, p_end_time);
    IF v_count = 0 THEN
        v_result := v_result || '(no changes detected)' || E'\n\n';
    ELSE
        v_result := v_result || '| Parameter | Old Value | New Value | Old Source | New Source | Changed At |' || E'\n';
        v_result := v_result || '|-----------|-----------|-----------|------------|------------|------------|' || E'\n';
        FOR v_row IN SELECT * FROM pgfr_analyze.config_changes(p_start_time, p_end_time) LOOP
            v_result := v_result || '| ' ||
                COALESCE(v_row.parameter_name, '-') || ' | ' ||
                COALESCE(v_row.old_value, '-') || ' | ' ||
                COALESCE(v_row.new_value, '-') || ' | ' ||
                COALESCE(v_row.old_source, '-') || ' | ' ||
                COALESCE(v_row.new_source, '-') || ' | ' ||
                COALESCE(to_char(v_row.changed_at, 'YYYY-MM-DD HH24:MI:SS'), '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- ==========================================================================
    -- Role Configuration Changes Section
    -- ==========================================================================
    v_result := v_result || '## Role Configuration Changes' || E'\n\n';

    SELECT count(*) INTO v_count FROM pgfr_analyze.db_role_config_changes(p_start_time, p_end_time);
    IF v_count = 0 THEN
        v_result := v_result || '(no changes detected)' || E'\n\n';
    ELSE
        v_result := v_result || '| Database | Role | Parameter | Old Value | New Value | Type |' || E'\n';
        v_result := v_result || '|----------|------|-----------|-----------|-----------|------|' || E'\n';
        FOR v_row IN SELECT * FROM pgfr_analyze.db_role_config_changes(p_start_time, p_end_time) LOOP
            v_result := v_result || '| ' ||
                COALESCE(v_row.database_name, '-') || ' | ' ||
                COALESCE(v_row.role_name, '-') || ' | ' ||
                COALESCE(v_row.parameter_name, '-') || ' | ' ||
                COALESCE(v_row.old_value, '-') || ' | ' ||
                COALESCE(v_row.new_value, '-') || ' | ' ||
                COALESCE(v_row.change_type, '-') || ' |' || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    RETURN v_result;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.report(TIMESTAMPTZ, TIMESTAMPTZ) IS
'Generate diagnostic report from flight recorder data. Readable by humans and AI systems.';

-- Interval convenience overload: report('1 hour') instead of timestamps
CREATE OR REPLACE FUNCTION pgfr_analyze.report(
    p_interval INTERVAL
)
RETURNS TEXT
LANGUAGE sql STABLE AS $$
    SELECT pgfr_analyze.report(now() - p_interval, now());
$$;
COMMENT ON FUNCTION pgfr_analyze.report(INTERVAL) IS
'Generate diagnostic report for the specified interval ending now. Usage: SELECT pgfr_analyze.report(''1 hour'')';
-- Validates system readiness for flight recorder installation by checking resources, connections, and dependencies
-- Returns component status (GO/CAUTION/NO-GO) to determine installation viability
CREATE OR REPLACE FUNCTION pgfr_analyze.preflight_check()
RETURNS TABLE(
    check_name TEXT,
    status TEXT,
    details TEXT,
    recommendation TEXT
)
LANGUAGE plpgsql AS $$
DECLARE
    v_max_connections INTEGER;
    v_current_connections INTEGER;
    v_pg_stat_statements_max INTEGER;
    v_cpu_count INTEGER;
    v_shared_buffers_mb INTEGER;
    v_connection_pct NUMERIC;
    v_pg_cron_exists BOOLEAN;
BEGIN
    BEGIN
        v_cpu_count := (SELECT setting::integer FROM pg_settings WHERE name = 'max_worker_processes');
        IF v_cpu_count < 4 THEN
            RETURN QUERY SELECT
                'System Resources'::text,
                'CAUTION'::text,
                format('Detected %s worker processes (CPU estimate)', v_cpu_count),
                'Flight recorder overhead (0.5%%) is acceptable but consider testing in staging first. Systems with <4 cores have less overhead margin.'::text;
        ELSE
            RETURN QUERY SELECT
                'System Resources'::text,
                'GO'::text,
                format('Detected %s worker processes', v_cpu_count),
                'System has adequate resources for always-on monitoring.'::text;
        END IF;
    EXCEPTION WHEN OTHERS THEN
        RETURN QUERY SELECT
            'System Resources'::text,
            'GO'::text,
            'Unable to detect CPU count',
            'Verify your system has ≥4 CPU cores for comfortable always-on operation.'::text;
    END;
    SELECT setting::integer INTO v_max_connections FROM pg_settings WHERE name = 'max_connections';
    SELECT count(*) INTO v_current_connections FROM pg_stat_activity;
    v_connection_pct := (v_current_connections::numeric / v_max_connections) * 100;
    IF v_connection_pct > 70 THEN
        RETURN QUERY SELECT
            'Connection Headroom'::text,
            'CAUTION'::text,
            format('Currently %s%% of max_connections (%s/%s)', round(v_connection_pct, 1), v_current_connections, v_max_connections),
            'System frequently near max_connections. Adaptive mode will trigger often. Consider increasing max_connections or monitoring connection patterns.'::text;
    ELSE
        RETURN QUERY SELECT
            'Connection Headroom'::text,
            'GO'::text,
            format('Currently %s%% of max_connections (%s/%s)', round(v_connection_pct, 1), v_current_connections, v_max_connections),
            'Adequate connection headroom for normal operations.'::text;
    END IF;
    BEGIN
        SELECT setting::integer INTO v_pg_stat_statements_max
        FROM pg_settings WHERE name = 'pg_stat_statements.max';
        IF v_pg_stat_statements_max < 5000 THEN
            RETURN QUERY SELECT
                'pg_stat_statements Budget'::text,
                'NO-GO'::text,
                format('pg_stat_statements.max = %s (too low)', v_pg_stat_statements_max),
                'Flight recorder will consume 20-40% of statement budget. Increase to at least 10,000: ALTER SYSTEM SET pg_stat_statements.max = 10000; (requires restart)'::text;
        ELSIF v_pg_stat_statements_max < 10000 THEN
            RETURN QUERY SELECT
                'pg_stat_statements Budget'::text,
                'CAUTION'::text,
                format('pg_stat_statements.max = %s (minimal)', v_pg_stat_statements_max),
                'Flight recorder will consume 20-40% of statement budget. Consider increasing to 10,000+ for comfortable operation.'::text;
        ELSE
            RETURN QUERY SELECT
                'pg_stat_statements Budget'::text,
                'GO'::text,
                format('pg_stat_statements.max = %s', v_pg_stat_statements_max),
                'Adequate statement tracking capacity.'::text;
        END IF;
    EXCEPTION WHEN OTHERS THEN
        RETURN QUERY SELECT
            'pg_stat_statements Budget'::text,
            'CAUTION'::text,
            'pg_stat_statements extension not found or not configured',
            'Statement collection will be disabled (statements_enabled=auto). This is acceptable if you don''t need query-level metrics.'::text;
    END;
    BEGIN
        SELECT setting::integer INTO v_shared_buffers_mb
        FROM pg_settings WHERE name = 'shared_buffers';
        v_shared_buffers_mb := (v_shared_buffers_mb * 8) / 1024;
        RETURN QUERY SELECT
            'Storage Overhead'::text,
            'GO'::text,
            'Ring buffer memory depends on slot count (default 120 slots, configurable 72-2880). Aggregates: ~2-3 GB per week (7-day retention).',
            'UNLOGGED ring buffers minimize WAL overhead. Ring buffers self-clean automatically. Daily aggregate cleanup prevents unbounded growth.'::text;
    END;
    SELECT EXISTS (
        SELECT 1 FROM pg_extension WHERE extname = 'pg_cron'
    ) INTO v_pg_cron_exists;
    IF v_pg_cron_exists THEN
        RETURN QUERY SELECT
            'Scheduling (pg_cron)'::text,
            'GO'::text,
            'pg_cron extension detected',
            'Automatic scheduling will work. Flight recorder will run automatically every minute.'::text;
    ELSE
        RETURN QUERY SELECT
            'Scheduling (pg_cron)'::text,
            'CAUTION'::text,
            'pg_cron extension not found',
            'You will need to schedule pgfr_record.sample() and pgfr_record.snapshot() manually via external cron or pg_agent.'::text;
    END IF;
    RETURN QUERY SELECT
        'Safety Mechanisms'::text,
        'GO'::text,
        'Circuit breaker, adaptive mode, timeouts all enabled by default',
        'Flight recorder will auto-reduce overhead under stress.'::text;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.preflight_check() IS

'Pre-installation validation checks. Returns component status (GO/CAUTION/NO-GO). For summary, use preflight_check_with_summary().';
-- Executes preflight validation checks and appends a summary row indicating overall system readiness (READY, CAUTION, or NO-GO)
CREATE OR REPLACE FUNCTION pgfr_analyze.preflight_check_with_summary()
RETURNS TABLE(
    check_name TEXT,
    status TEXT,
    details TEXT,
    recommendation TEXT
)
LANGUAGE plpgsql AS $$
DECLARE
    v_nogo_count INTEGER;
    v_caution_count INTEGER;
BEGIN
    RETURN QUERY SELECT * FROM pgfr_analyze.preflight_check();
    SELECT
        count(*) FILTER (WHERE c.status = 'NO-GO'),
        count(*) FILTER (WHERE c.status = 'CAUTION')
    INTO v_nogo_count, v_caution_count
    FROM pgfr_analyze.preflight_check() c;
    IF v_nogo_count > 0 THEN
        RETURN QUERY SELECT
            '=== SUMMARY ==='::text,
            'NO-GO'::text,
            format('%s critical issues detected', v_nogo_count),
            'Address NO-GO items before enabling always-on monitoring. See recommendations above.'::text;
    ELSIF v_caution_count > 0 THEN
        RETURN QUERY SELECT
            '=== SUMMARY ==='::text,
            'PROCEED WITH CAUTION'::text,
            format('%s cautions detected', v_caution_count),
            'System is acceptable for always-on monitoring but consider addressing cautions. Test in staging first if possible.'::text;
    ELSE
        RETURN QUERY SELECT
            '=== SUMMARY ==='::text,
            'READY FOR PRODUCTION'::text,
            'All checks passed',
            'System is well-suited for "set and forget" always-on monitoring. Run quarterly_review() every 3 months to verify continued health.'::text;
    END IF;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.preflight_check_with_summary() IS

'Pre-installation validation with summary. Calls preflight_check() twice - once for results, once to count. More expensive but includes summary row.';
-- Generates a comprehensive quarterly health review of the pgfr_record system
-- Assesses collection performance, storage consumption, reliability, circuit breaker activity, and data freshness
CREATE OR REPLACE FUNCTION pgfr_analyze.quarterly_review()
RETURNS TABLE(
    component TEXT,
    status TEXT,
    metric TEXT,
    assessment TEXT
)
LANGUAGE plpgsql AS $$
DECLARE
    v_avg_duration_ms NUMERIC;
    v_max_duration_ms NUMERIC;
    v_skipped_count INTEGER;
    v_total_count INTEGER;
    v_schema_size_mb NUMERIC;
    v_last_sample TIMESTAMPTZ;
    v_last_snapshot TIMESTAMPTZ;
    v_circuit_breaker_trips INTEGER;
    v_failed_collections INTEGER;
    v_sample_count INTEGER;
BEGIN
    RETURN QUERY SELECT
        '=== FLIGHT RECORDER QUARTERLY REVIEW ==='::text,
        'INFO'::text,
        format('Review period: Last 90 days | Generated: %s', now()::text),
        'This review validates flight recorder health for continued always-on operation.'::text;
    SELECT
        avg(duration_ms),
        max(duration_ms),
        count(*) FILTER (WHERE skipped),
        count(*)
    INTO v_avg_duration_ms, v_max_duration_ms, v_skipped_count, v_total_count
    FROM pgfr_record.collection_stats
    WHERE started_at > now() - interval '30 days'
      AND collection_type = 'sample';
    IF v_avg_duration_ms IS NULL THEN
        RETURN QUERY SELECT
            '1. Collection Performance'::text,
            'ERROR'::text,
            'No collections in last 30 days',
            'Flight recorder may not be running. Check pg_cron jobs.'::text;
    ELSIF v_avg_duration_ms < 200 THEN
        RETURN QUERY SELECT
            '1. Collection Performance'::text,
            'EXCELLENT'::text,
            format('Avg: %sms | Max: %sms | Skipped: %s/%s',
                   round(v_avg_duration_ms), round(v_max_duration_ms), v_skipped_count, v_total_count),
            'Collection overhead is minimal. No action needed.'::text;
    ELSIF v_avg_duration_ms < 500 THEN
        RETURN QUERY SELECT
            '1. Collection Performance'::text,
            'GOOD'::text,
            format('Avg: %sms | Max: %sms | Skipped: %s/%s',
                   round(v_avg_duration_ms), round(v_max_duration_ms), v_skipped_count, v_total_count),
            'Collection overhead is acceptable. Continue monitoring.'::text;
    ELSE
        RETURN QUERY SELECT
            '1. Collection Performance'::text,
            'REVIEW NEEDED'::text,
            format('Avg: %sms | Max: %sms | Skipped: %s/%s',
                   round(v_avg_duration_ms), round(v_max_duration_ms), v_skipped_count, v_total_count),
            'Collections are slower than expected. Consider: (1) switching to light mode, (2) increasing sample_interval_seconds to 300, or (3) checking for system bottlenecks.'::text;
    END IF;
    SELECT schema_size_mb INTO v_schema_size_mb FROM pgfr_record._check_schema_size();
    SELECT count(*) INTO v_sample_count FROM pgfr_record.samples_ring;
    IF v_schema_size_mb < 3000 THEN
        RETURN QUERY SELECT
            '2. Storage Consumption'::text,
            'EXCELLENT'::text,
            format('%s MB | %s samples', round(v_schema_size_mb), v_sample_count),
            'Storage usage is healthy. Daily cleanup is working correctly.'::text;
    ELSIF v_schema_size_mb < 6000 THEN
        RETURN QUERY SELECT
            '2. Storage Consumption'::text,
            'GOOD'::text,
            format('%s MB | %s samples', round(v_schema_size_mb), v_sample_count),
            'Storage usage is acceptable. Monitor growth trend.'::text;
    ELSE
        RETURN QUERY SELECT
            '2. Storage Consumption'::text,
            'REVIEW NEEDED'::text,
            format('%s MB | %s samples', round(v_schema_size_mb), v_sample_count),
            'Storage usage is high. Run cleanup() or reduce retention_samples_days.'::text;
    END IF;
    SELECT
        count(*) FILTER (WHERE NOT success AND NOT skipped)
    INTO v_failed_collections
    FROM pgfr_record.collection_stats
    WHERE started_at > now() - interval '90 days';
    IF v_failed_collections = 0 THEN
        RETURN QUERY SELECT
            '3. Collection Reliability'::text,
            'EXCELLENT'::text,
            format('0 failed collections in 90 days'),
            'Flight recorder is operating reliably.'::text;
    ELSIF v_failed_collections < 10 THEN
        RETURN QUERY SELECT
            '3. Collection Reliability'::text,
            'GOOD'::text,
            format('%s failed collections in 90 days', v_failed_collections),
            'Minimal failures detected. This is normal and acceptable.'::text;
    ELSE
        RETURN QUERY SELECT
            '3. Collection Reliability'::text,
            'REVIEW NEEDED'::text,
            format('%s failed collections in 90 days', v_failed_collections),
            'Frequent failures detected. Check collection_stats for error patterns.'::text;
    END IF;
    SELECT count(*) INTO v_circuit_breaker_trips
    FROM pgfr_record.collection_stats
    WHERE skipped = true
      AND skipped_reason LIKE '%Circuit breaker%'
      AND started_at > now() - interval '90 days';
    IF v_circuit_breaker_trips = 0 THEN
        RETURN QUERY SELECT
            '4. Circuit Breaker Activity'::text,
            'EXCELLENT'::text,
            '0 trips in 90 days',
            'System has not experienced collection stress.'::text;
    ELSIF v_circuit_breaker_trips < 20 THEN
        RETURN QUERY SELECT
            '4. Circuit Breaker Activity'::text,
            'GOOD'::text,
            format('%s trips in 90 days', v_circuit_breaker_trips),
            'Circuit breaker has triggered occasionally. This is normal during temporary load spikes.'::text;
    ELSE
        RETURN QUERY SELECT
            '4. Circuit Breaker Activity'::text,
            'REVIEW NEEDED'::text,
            format('%s trips in 90 days', v_circuit_breaker_trips),
            'Frequent circuit breaker trips indicate system stress. Consider switching to light mode permanently.'::text;
    END IF;
    SELECT max(captured_at) INTO v_last_sample FROM pgfr_record.samples_ring;
    SELECT max(captured_at) INTO v_last_snapshot FROM pgfr_record.snapshots;
    IF v_last_sample > now() - interval '10 minutes' AND v_last_snapshot > now() - interval '15 minutes' THEN
        RETURN QUERY SELECT
            '5. Data Freshness'::text,
            'EXCELLENT'::text,
            format('Last sample: %s ago | Last snapshot: %s ago',
                   age(now(), v_last_sample)::text, age(now(), v_last_snapshot)::text),
            'Collections are running on schedule.'::text;
    ELSE
        RETURN QUERY SELECT
            '5. Data Freshness'::text,
            'ERROR'::text,
            format('Last sample: %s ago | Last snapshot: %s ago',
                   age(now(), v_last_sample)::text, age(now(), v_last_snapshot)::text),
            'Collections are stale. Check pg_cron jobs: SELECT * FROM cron.job WHERE jobname LIKE ''pgfr_%'';'::text;
    END IF;
    DECLARE
        v_missing_count INTEGER;
        v_inactive_count INTEGER;
        v_missing_jobs TEXT[];
        v_inactive_jobs TEXT[];
    BEGIN
        WITH required_jobs AS (
            SELECT unnest(ARRAY[
                'pgfr_sample',
                'pgfr_snapshot',
                'pgfr_flush',
                'pgfr_cleanup'
            ]) AS job_name
        )
        SELECT
            count(*) FILTER (WHERE j.jobid IS NULL),
            count(*) FILTER (WHERE j.jobid IS NOT NULL AND NOT j.active),
            array_agg(r.job_name) FILTER (WHERE j.jobid IS NULL),
            array_agg(r.job_name) FILTER (WHERE j.jobid IS NOT NULL AND NOT j.active)
        INTO v_missing_count, v_inactive_count, v_missing_jobs, v_inactive_jobs
        FROM required_jobs r
        LEFT JOIN cron.job j ON j.jobname = r.job_name;
        IF v_missing_count = 0 AND v_inactive_count = 0 THEN
            RETURN QUERY SELECT
                '6. pg_cron Job Health'::text,
                'EXCELLENT'::text,
                '4/4 jobs active (sample, snapshot, flush, cleanup)',
                'All pg_cron jobs are running correctly.'::text;
        ELSIF v_missing_count > 0 THEN
            RETURN QUERY SELECT
                '6. pg_cron Job Health'::text,
                'CRITICAL'::text,
                format('%s/%s jobs missing: %s', v_missing_count, 4, array_to_string(v_missing_jobs, ', ')),
                'CRITICAL: Flight recorder is not collecting data. Run pgfr_record.enable() to restore.'::text;
        ELSE
            RETURN QUERY SELECT
                '6. pg_cron Job Health'::text,
                'CRITICAL'::text,
                format('%s/%s jobs inactive: %s', v_inactive_count, 4, array_to_string(v_inactive_jobs, ', ')),
                'CRITICAL: pg_cron jobs exist but are disabled. Run pgfr_record.enable() to reactivate.'::text;
        END IF;
    EXCEPTION WHEN OTHERS THEN
        RETURN QUERY SELECT
            '7. pg_cron Job Health'::text,
            'ERROR'::text,
            format('Failed to check pg_cron jobs: %s', SQLERRM),
            'Verify pg_cron extension is installed and accessible.'::text;
    END;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.quarterly_review() IS

'Quarterly health check for flight recorder. Returns component metrics (EXCELLENT/GOOD/REVIEW NEEDED/ERROR). For summary, use quarterly_review_with_summary().';
-- Quarterly health check with summary. Appends overall health status (HEALTHY or ACTION REQUIRED) based on count of ERROR or REVIEW NEEDED items detected
-- More expensive than quarterly_review() as it calls it twice - once for detailed results, once to count critical issues
CREATE OR REPLACE FUNCTION pgfr_analyze.quarterly_review_with_summary()
RETURNS TABLE(
    component TEXT,
    status TEXT,
    metric TEXT,
    assessment TEXT
)
LANGUAGE plpgsql AS $$
DECLARE
    v_issues_count INTEGER;
BEGIN
    RETURN QUERY SELECT * FROM pgfr_analyze.quarterly_review();
    SELECT count(*) INTO v_issues_count
    FROM pgfr_analyze.quarterly_review() qr
    WHERE qr.status IN ('ERROR', 'REVIEW NEEDED');
    IF v_issues_count = 0 THEN
        RETURN QUERY SELECT
            '=== QUARTERLY REVIEW SUMMARY ==='::text,
            'HEALTHY'::text,
            'All metrics within acceptable parameters',
            'Flight recorder is operating as expected. Schedule next review in 3 months. No action required.'::text;
    ELSE
        RETURN QUERY SELECT
            '=== QUARTERLY REVIEW SUMMARY ==='::text,
            'ACTION REQUIRED'::text,
            format('%s items need review', v_issues_count),
            'Address items marked ERROR or REVIEW NEEDED above. Run config_recommendations() for specific tuning suggestions.'::text;
    END IF;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.quarterly_review_with_summary() IS

'Quarterly health check with summary. Calls quarterly_review() twice - once for results, once to count. More expensive but includes summary row.';
-- Analyzes database resource capacity metrics (connections, memory, storage, transactions) over a time window
-- Returns utilization status with actionable recommendations

-- ==========================================================
-- pgfr_analyze/sql/05_capacity.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

CREATE OR REPLACE FUNCTION pgfr_analyze.capacity_summary(
    p_time_window INTERVAL DEFAULT interval '24 hours'
)
RETURNS TABLE(
    metric                  TEXT,
    current_usage           TEXT,
    provisioned_capacity    TEXT,
    utilization_pct         NUMERIC,
    headroom_pct            NUMERIC,
    status                  TEXT,
    recommendation          TEXT
)
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_warning_pct INTEGER;
    v_critical_pct INTEGER;
    v_window_start TIMESTAMPTZ;
    v_current_connections INTEGER;
    v_max_connections INTEGER;
    v_avg_connections NUMERIC;
    v_peak_connections INTEGER;
    v_bgw_backend_total BIGINT;
    v_bgw_backend_avg_per_min NUMERIC;
    v_shared_buffers_setting TEXT;
    v_temp_bytes_total BIGINT;
    v_temp_bytes_per_hour NUMERIC;
    v_blks_read_total BIGINT;
    v_blks_hit_total BIGINT;
    v_cache_hit_ratio NUMERIC;
    v_current_db_size BIGINT;
    v_oldest_db_size BIGINT;
    v_storage_growth_mb_per_day NUMERIC;
    v_xact_total BIGINT;
    v_xact_rate_avg NUMERIC;
    v_xact_rate_peak NUMERIC;
    v_window_hours NUMERIC;
    v_sample_count INTEGER;
BEGIN
    v_warning_pct := COALESCE(
        pgfr_record._get_config('capacity_thresholds_warning_pct', '60')::integer,
        60
    );
    v_critical_pct := COALESCE(
        pgfr_record._get_config('capacity_thresholds_critical_pct', '80')::integer,
        80
    );
    v_window_start := now() - p_time_window;
    v_window_hours := EXTRACT(EPOCH FROM p_time_window) / 3600.0;
    SELECT count(*) INTO v_sample_count
    FROM pgfr_record.snapshots
    WHERE captured_at >= v_window_start;
    IF v_sample_count < 2 THEN
        RETURN QUERY SELECT
            'insufficient_data'::text,
            NULL::text,
            NULL::text,
            NULL::numeric,
            NULL::numeric,
            'insufficient_data'::text,
            format('Need at least 2 snapshots. Only %s found in window. Wait %s for capacity analysis.',
                   v_sample_count,
                   CASE WHEN v_sample_count = 0 THEN '5 minutes' ELSE 'a few more minutes' END)::text;
        RETURN;
    END IF;
    SELECT setting::integer INTO v_max_connections
    FROM pg_settings WHERE name = 'max_connections';
    SELECT COALESCE(connections_total, 0)
    INTO v_current_connections
    FROM pgfr_record.snapshots
    WHERE captured_at >= v_window_start
      AND connections_total IS NOT NULL
    ORDER BY captured_at DESC
    LIMIT 1;
    SELECT
        COALESCE(round(avg(connections_total), 0), 0),
        COALESCE(max(connections_total), 0)
    INTO v_avg_connections, v_peak_connections
    FROM pgfr_record.snapshots
    WHERE captured_at >= v_window_start
      AND connections_total IS NOT NULL;
    IF v_current_connections IS NOT NULL THEN
        metric := 'connections';
        current_usage := format('%s current / %s avg / %s peak',
                               v_current_connections,
                               v_avg_connections::integer,
                               v_peak_connections);
        provisioned_capacity := v_max_connections::text;
        utilization_pct := LEAST(100, round((v_peak_connections::numeric / NULLIF(v_max_connections, 0)) * 100, 1));
        headroom_pct := CASE WHEN utilization_pct IS NOT NULL THEN round(GREATEST(0, 100 - utilization_pct), 1) ELSE NULL END;
        status := CASE
            WHEN utilization_pct >= v_critical_pct THEN 'critical'
            WHEN utilization_pct >= v_warning_pct THEN 'warning'
            ELSE 'healthy'
        END;
        recommendation := CASE
            WHEN utilization_pct >= v_critical_pct THEN
                format('CRITICAL: Peak connections at %s%%. Increase max_connections to %s+ or implement connection pooling (PgBouncer)',
                       utilization_pct, (v_peak_connections * 1.5)::integer)
            WHEN utilization_pct >= v_warning_pct THEN
                format('WARNING: Peak connections at %s%%. Monitor closely. Consider connection pooling if trend continues.',
                       utilization_pct)
            WHEN utilization_pct < 40 THEN
                format('HEALTHY: Peak usage %s%% (%s connections). max_connections may be over-provisioned (potential cost savings).',
                       utilization_pct, v_peak_connections)
            ELSE
                format('HEALTHY: Peak usage %s%% with %s%% headroom. No action needed.',
                       utilization_pct, headroom_pct)
        END;
        RETURN NEXT;
    END IF;
    WITH buffer_deltas AS (
        SELECT
            s.captured_at,
            s.bgw_buffers_backend - prev.bgw_buffers_backend AS backend_writes_delta,
            EXTRACT(EPOCH FROM (s.captured_at - prev.captured_at)) / 60.0 AS interval_minutes
        FROM pgfr_record.snapshots s
        JOIN pgfr_record.snapshots prev ON prev.id = (
            SELECT MAX(id) FROM pgfr_record.snapshots WHERE id < s.id
        )
        WHERE s.captured_at >= v_window_start
          AND s.bgw_buffers_backend IS NOT NULL
          AND prev.bgw_buffers_backend IS NOT NULL
          AND s.bgw_buffers_backend >= prev.bgw_buffers_backend
    )
    SELECT
        GREATEST(0, COALESCE(sum(backend_writes_delta), 0)),
        GREATEST(0, COALESCE(sum(backend_writes_delta) / NULLIF(sum(interval_minutes), 0), 0))
    INTO v_bgw_backend_total, v_bgw_backend_avg_per_min
    FROM buffer_deltas;
    SELECT setting INTO v_shared_buffers_setting
    FROM pg_settings WHERE name = 'shared_buffers';
    metric := 'memory_shared_buffers';
    current_usage := format('%s backend writes total (%s/min avg)',
                           v_bgw_backend_total,
                           round(v_bgw_backend_avg_per_min, 1));
    provisioned_capacity := v_shared_buffers_setting;
    utilization_pct := CASE
        WHEN v_bgw_backend_avg_per_min IS NULL OR v_bgw_backend_avg_per_min <= 0 THEN 0
        WHEN v_bgw_backend_avg_per_min < 100 THEN round((v_bgw_backend_avg_per_min / 100.0) * 60, 1)
        ELSE LEAST(100, round(60 + ((v_bgw_backend_avg_per_min - 100) / 900.0) * 40, 1))
    END;
    headroom_pct := round(GREATEST(0, 100 - utilization_pct), 1);
    status := CASE
        WHEN v_bgw_backend_avg_per_min >= 1000 THEN 'critical'
        WHEN v_bgw_backend_avg_per_min >= 100 THEN 'warning'
        ELSE 'healthy'
    END;
    recommendation := CASE
        WHEN v_bgw_backend_avg_per_min >= 1000 THEN
            format('CRITICAL: Heavy shared_buffers pressure (%s backend writes/min). Increase shared_buffers or reduce concurrent write load.',
                   round(v_bgw_backend_avg_per_min, 1))
        WHEN v_bgw_backend_avg_per_min >= 100 THEN
            format('WARNING: Moderate shared_buffers pressure (%s backend writes/min). Monitor trend and consider increasing shared_buffers.',
                   round(v_bgw_backend_avg_per_min, 1))
        WHEN v_bgw_backend_total = 0 THEN
            'HEALTHY: No shared_buffers pressure detected. Current setting appears adequate.'
        ELSE
            format('HEALTHY: Minimal shared_buffers pressure (%s backend writes/min). No action needed.',
                   round(v_bgw_backend_avg_per_min, 1))
    END;
    RETURN NEXT;
    WITH temp_deltas AS (
        SELECT
            s.temp_bytes - prev.temp_bytes AS temp_bytes_delta
        FROM pgfr_record.snapshots s
        JOIN pgfr_record.snapshots prev ON prev.id = (
            SELECT MAX(id) FROM pgfr_record.snapshots WHERE id < s.id
        )
        WHERE s.captured_at >= v_window_start
          AND s.temp_bytes IS NOT NULL
          AND prev.temp_bytes IS NOT NULL
          AND s.temp_bytes >= prev.temp_bytes
    )
    SELECT
        COALESCE(sum(temp_bytes_delta), 0)
    INTO v_temp_bytes_total
    FROM temp_deltas;
    v_temp_bytes_per_hour := v_temp_bytes_total / NULLIF(v_window_hours, 0);
    metric := 'memory_work_mem';
    current_usage := format('%s spilled (%s/hour)',
                           pgfr_record._pretty_bytes(v_temp_bytes_total),
                           pgfr_record._pretty_bytes(v_temp_bytes_per_hour::bigint));
    provisioned_capacity := (SELECT setting FROM pg_settings WHERE name = 'work_mem');
    utilization_pct := CASE
        WHEN v_temp_bytes_per_hour IS NULL OR v_temp_bytes_per_hour <= 0 THEN 0
        WHEN v_temp_bytes_per_hour < 104857600 THEN round((v_temp_bytes_per_hour / 104857600.0) * 60, 1)
        ELSE LEAST(100, round(60 + ((v_temp_bytes_per_hour - 104857600) / 939524096.0) * 40, 1))
    END;
    headroom_pct := round(GREATEST(0, 100 - utilization_pct), 1);
    status := CASE
        WHEN v_temp_bytes_per_hour >= 1073741824 THEN 'critical'
        WHEN v_temp_bytes_per_hour >= 104857600 THEN 'warning'
        ELSE 'healthy'
    END;
    recommendation := CASE
        WHEN v_temp_bytes_per_hour >= 1073741824 THEN
            format('CRITICAL: Heavy temp file spills (%s/hour). Increase work_mem for affected queries or globally.',
                   pgfr_record._pretty_bytes(v_temp_bytes_per_hour::bigint))
        WHEN v_temp_bytes_per_hour >= 104857600 THEN
            format('WARNING: Moderate temp file spills (%s/hour). Consider increasing work_mem for sort/hash operations.',
                   pgfr_record._pretty_bytes(v_temp_bytes_per_hour::bigint))
        WHEN v_temp_bytes_total = 0 THEN
            'HEALTHY: No temp file spills detected. Queries fitting in work_mem.'
        ELSE
            format('HEALTHY: Minimal temp file spills (%s/hour). Current work_mem adequate.',
                   pgfr_record._pretty_bytes(v_temp_bytes_per_hour::bigint))
    END;
    RETURN NEXT;
    WITH io_deltas AS (
        SELECT
            s.blks_read - prev.blks_read AS read_delta,
            s.blks_hit - prev.blks_hit AS hit_delta
        FROM pgfr_record.snapshots s
        JOIN pgfr_record.snapshots prev ON prev.id = (
            SELECT MAX(id) FROM pgfr_record.snapshots WHERE id < s.id
        )
        WHERE s.captured_at >= v_window_start
          AND s.blks_read IS NOT NULL
          AND prev.blks_read IS NOT NULL
          AND s.blks_read >= prev.blks_read
    )
    SELECT
        COALESCE(sum(read_delta), 0),
        COALESCE(sum(hit_delta), 0)
    INTO v_blks_read_total, v_blks_hit_total
    FROM io_deltas;
    v_cache_hit_ratio := CASE
        WHEN (v_blks_read_total + v_blks_hit_total) > 0
        THEN round((v_blks_hit_total::numeric / (v_blks_read_total + v_blks_hit_total)) * 100, 2)
        ELSE NULL
    END;
    metric := 'io_buffer_cache';
    current_usage := format('%s%% cache hit ratio (%s reads / %s hits)',
                           v_cache_hit_ratio,
                           v_blks_read_total,
                           v_blks_hit_total);
    provisioned_capacity := 'Target: >95%';
    utilization_pct := CASE
        WHEN v_cache_hit_ratio IS NULL THEN NULL
        WHEN v_cache_hit_ratio >= 95 THEN LEAST(100, GREATEST(0, round((100 - v_cache_hit_ratio) * 20, 1)))
        ELSE LEAST(100, GREATEST(0, round(100 - (v_cache_hit_ratio / 95.0) * 100, 1)))
    END;
    headroom_pct := CASE WHEN utilization_pct IS NOT NULL THEN round(GREATEST(0, 100 - utilization_pct), 1) ELSE NULL END;
    status := CASE
        WHEN v_cache_hit_ratio IS NULL THEN 'insufficient_data'
        WHEN v_cache_hit_ratio < 80 THEN 'critical'
        WHEN v_cache_hit_ratio < 95 THEN 'warning'
        ELSE 'healthy'
    END;
    recommendation := CASE
        WHEN v_cache_hit_ratio IS NULL THEN
            'Insufficient I/O data. Need more snapshots for cache hit ratio analysis.'
        WHEN v_cache_hit_ratio < 80 THEN
            format('CRITICAL: Poor cache hit ratio (%s%%). Increase shared_buffers or optimize queries to reduce I/O.',
                   v_cache_hit_ratio)
        WHEN v_cache_hit_ratio < 95 THEN
            format('WARNING: Below-optimal cache hit ratio (%s%%). Consider increasing shared_buffers for better performance.',
                   v_cache_hit_ratio)
        ELSE
            format('HEALTHY: Good cache hit ratio (%s%%). I/O performance is adequate.',
                   v_cache_hit_ratio)
    END;
    RETURN NEXT;
    SELECT
        COALESCE(
            (SELECT db_size_bytes FROM pgfr_record.snapshots
             WHERE captured_at >= v_window_start AND db_size_bytes IS NOT NULL
             ORDER BY captured_at DESC LIMIT 1),
            0
        ),
        COALESCE(
            (SELECT db_size_bytes FROM pgfr_record.snapshots
             WHERE captured_at >= v_window_start AND db_size_bytes IS NOT NULL
             ORDER BY captured_at ASC LIMIT 1),
            0
        )
    INTO v_current_db_size, v_oldest_db_size;
    IF v_current_db_size > 0 AND v_oldest_db_size > 0 THEN
        v_storage_growth_mb_per_day := ((v_current_db_size - v_oldest_db_size)::numeric / (1024.0 * 1024.0))
                                       / NULLIF(v_window_hours / 24.0, 0);
        metric := 'storage_growth';
        current_usage := format('%s current size, growing %s MB/day',
                               pgfr_record._pretty_bytes(v_current_db_size),
                               CASE
                                   WHEN v_storage_growth_mb_per_day < 0 THEN '~0'
                                   ELSE round(v_storage_growth_mb_per_day, 1)::text
                               END);
        provisioned_capacity := 'Disk capacity dependent';
        utilization_pct := CASE
            WHEN v_storage_growth_mb_per_day <= 0 THEN 0
            WHEN v_storage_growth_mb_per_day < 1024 THEN round((v_storage_growth_mb_per_day / 1024.0) * 60, 1)
            ELSE LEAST(100, round(60 + ((v_storage_growth_mb_per_day - 1024) / 9216.0) * 40, 1))
        END;
        headroom_pct := round(GREATEST(0, 100 - utilization_pct), 1);
        status := CASE
            WHEN v_storage_growth_mb_per_day >= 10240 THEN 'critical'
            WHEN v_storage_growth_mb_per_day >= 1024 THEN 'warning'
            ELSE 'healthy'
        END;
        recommendation := CASE
            WHEN v_storage_growth_mb_per_day >= 10240 THEN
                format('CRITICAL: Rapid storage growth (%s MB/day). Review VACUUM, bloat, and retention policies.',
                       round(v_storage_growth_mb_per_day, 1))
            WHEN v_storage_growth_mb_per_day >= 1024 THEN
                format('WARNING: Significant storage growth (%s MB/day). Monitor disk capacity and plan expansion.',
                       round(v_storage_growth_mb_per_day, 1))
            WHEN v_storage_growth_mb_per_day < 0 THEN
                'HEALTHY: Database size stable or shrinking (VACUUM/DELETE activity). No concerns.'
            ELSE
                format('HEALTHY: Moderate storage growth (%s MB/day). Current rate is sustainable.',
                       round(v_storage_growth_mb_per_day, 1))
        END;
        RETURN NEXT;
    END IF;
    WITH xact_deltas AS (
        SELECT
            (s.xact_commit + s.xact_rollback - prev.xact_commit - prev.xact_rollback) AS xact_delta,
            EXTRACT(EPOCH FROM (s.captured_at - prev.captured_at)) AS interval_seconds
        FROM pgfr_record.snapshots s
        JOIN pgfr_record.snapshots prev ON prev.id = (
            SELECT MAX(id) FROM pgfr_record.snapshots WHERE id < s.id
        )
        WHERE s.captured_at >= v_window_start
          AND s.xact_commit IS NOT NULL
          AND prev.xact_commit IS NOT NULL
          AND (s.xact_commit + s.xact_rollback) >= (prev.xact_commit + prev.xact_rollback)
    )
    SELECT
        COALESCE(sum(xact_delta), 0),
        COALESCE(round(avg(xact_delta / NULLIF(interval_seconds, 0)), 1), 0),
        COALESCE(round(max(xact_delta / NULLIF(interval_seconds, 0)), 1), 0)
    INTO v_xact_total, v_xact_rate_avg, v_xact_rate_peak
    FROM xact_deltas;
    IF v_xact_total > 0 THEN
        metric := 'transaction_rate';
        current_usage := format('%s total (%s/sec avg, %s/sec peak)',
                               v_xact_total,
                               v_xact_rate_avg,
                               v_xact_rate_peak);
        provisioned_capacity := 'Workload dependent';
        utilization_pct := CASE
            WHEN v_xact_rate_peak IS NULL OR v_xact_rate_peak <= 0 THEN 0
            WHEN v_xact_rate_peak < 1000 THEN round((v_xact_rate_peak / 1000.0) * 60, 1)
            ELSE LEAST(100, round(60 + ((v_xact_rate_peak - 1000) / 4000.0) * 40, 1))
        END;
        headroom_pct := round(GREATEST(0, 100 - utilization_pct), 1);
        status := CASE
            WHEN v_xact_rate_peak >= 5000 THEN 'warning'
            ELSE 'healthy'
        END;
        recommendation := CASE
            WHEN v_xact_rate_peak >= 5000 THEN
                format('High transaction rate (%s tps peak). Ensure connection pooling and monitoring CPU/I/O capacity.',
                       v_xact_rate_peak)
            WHEN v_xact_rate_avg < 10 THEN
                format('Low transaction rate (%s tps avg). Database may be under-utilized.',
                       v_xact_rate_avg)
            ELSE
                format('HEALTHY: Transaction rate %s tps avg, %s tps peak. Workload appears normal.',
                       v_xact_rate_avg, v_xact_rate_peak)
        END;
        RETURN NEXT;
    END IF;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.capacity_summary(INTERVAL) IS

'Capacity planning summary across all resource dimensions. Analyzes connections, memory (shared_buffers, work_mem), I/O (cache hit ratio), storage growth, and transaction rates over the specified time window. Returns utilization, status (healthy/warning/critical), and actionable recommendations. Part of Phase 1 MVP capacity planning enhancements (FR-2.1).';

-- Generates a human-readable markdown capacity report grouped by severity
-- Calls capacity_summary() and formats results as prose text
CREATE OR REPLACE FUNCTION pgfr_analyze.capacity_report(
    p_time_window INTERVAL DEFAULT interval '24 hours'
)
RETURNS TEXT
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_result TEXT := '';
    v_row RECORD;
    v_has_critical BOOLEAN := FALSE;
    v_has_warning BOOLEAN := FALSE;
    v_has_healthy BOOLEAN := FALSE;
    v_critical_section TEXT := '';
    v_warning_section TEXT := '';
    v_healthy_section TEXT := '';
    v_metric_name TEXT;
    v_recommendation TEXT;
BEGIN
    -- Header
    v_result := '# Capacity Report' || E'\n';
    v_result := v_result || 'Generated: ' || to_char(now(), 'YYYY-MM-DD HH24:MI:SS') ||
                ' | Window: ' || p_time_window::text || E'\n';

    -- Process each metric from capacity_summary
    FOR v_row IN
        SELECT *
        FROM pgfr_analyze.capacity_summary(p_time_window)
        WHERE metric != 'insufficient_data'
        ORDER BY
            CASE status
                WHEN 'critical' THEN 1
                WHEN 'warning' THEN 2
                ELSE 3
            END,
            metric
    LOOP
        -- Map metric names to readable names
        v_metric_name := CASE v_row.metric
            WHEN 'connections' THEN 'Connections'
            WHEN 'memory_shared_buffers' THEN 'Shared Buffers'
            WHEN 'memory_work_mem' THEN 'Work Mem'
            WHEN 'io_buffer_cache' THEN 'Cache Hit Ratio'
            WHEN 'storage_growth' THEN 'Storage Growth'
            WHEN 'transaction_rate' THEN 'Transaction Rate'
            ELSE initcap(replace(v_row.metric, '_', ' '))
        END;

        -- Strip severity prefix from recommendation
        v_recommendation := regexp_replace(v_row.recommendation, '^(CRITICAL|WARNING|HEALTHY): ', '');

        -- Build section content based on status
        IF v_row.status = 'critical' THEN
            v_has_critical := TRUE;
            v_critical_section := v_critical_section ||
                E'\n**' || v_metric_name || '** - ' || v_row.current_usage || E'\n' ||
                '  → ' || v_recommendation || E'\n';
        ELSIF v_row.status = 'warning' THEN
            v_has_warning := TRUE;
            v_warning_section := v_warning_section ||
                E'\n**' || v_metric_name || '** - ' || v_row.current_usage || E'\n' ||
                '  → ' || v_recommendation || E'\n';
        ELSE
            v_has_healthy := TRUE;
            v_healthy_section := v_healthy_section ||
                E'\n**' || v_metric_name || '** - ' || v_row.current_usage || E'\n';
        END IF;
    END LOOP;

    -- Handle insufficient data case
    FOR v_row IN
        SELECT *
        FROM pgfr_analyze.capacity_summary(p_time_window)
        WHERE metric = 'insufficient_data'
    LOOP
        v_result := v_result || E'\n' || v_row.recommendation || E'\n';
        RETURN v_result;
    END LOOP;

    -- Assemble sections
    IF v_has_critical THEN
        v_result := v_result || E'\n## Critical Issues' || v_critical_section;
    END IF;

    IF v_has_warning THEN
        v_result := v_result || E'\n## Warnings' || v_warning_section;
    END IF;

    IF v_has_healthy THEN
        v_result := v_result || E'\n## Healthy' || v_healthy_section;
    END IF;

    -- If nothing was found
    IF NOT v_has_critical AND NOT v_has_warning AND NOT v_has_healthy THEN
        v_result := v_result || E'\nNo capacity data available for the specified time window.\n';
    END IF;

    RETURN v_result;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.capacity_report(INTERVAL) IS
'Human-readable capacity report in markdown format. Groups metrics by severity (critical, warning, healthy) with actionable recommendations. Calls capacity_summary() internally for data. Use this for incident reports and documentation.';

CREATE OR REPLACE VIEW pgfr_analyze.capacity_dashboard AS
WITH
latest_snapshot AS (
    SELECT max(captured_at) AS last_updated
    FROM pgfr_record.snapshots
),
capacity_metrics AS (
    SELECT
        metric,
        utilization_pct,
        headroom_pct,
        status,
        recommendation
    FROM pgfr_analyze.capacity_summary(interval '24 hours')
    WHERE metric != 'insufficient_data'
),
connections_metric AS (
    SELECT
        status AS connections_status,
        utilization_pct AS connections_utilization_pct,
        headroom_pct AS connections_headroom,
        recommendation AS connections_recommendation
    FROM capacity_metrics
    WHERE metric = 'connections'
),
memory_sb_metric AS (
    SELECT
        status,
        utilization_pct
    FROM capacity_metrics
    WHERE metric = 'memory_shared_buffers'
),
memory_wm_metric AS (
    SELECT
        status,
        utilization_pct
    FROM capacity_metrics
    WHERE metric = 'memory_work_mem'
),
io_metric AS (
    SELECT
        status AS io_status,
        utilization_pct AS io_saturation_pct,
        recommendation AS io_recommendation
    FROM capacity_metrics
    WHERE metric = 'io_buffer_cache'
),
storage_metric AS (
    SELECT
        status AS storage_status,
        utilization_pct AS storage_utilization_pct,
        recommendation AS storage_recommendation
    FROM capacity_metrics
    WHERE metric = 'storage_growth'
)
SELECT
    l.last_updated,
    COALESCE(c.connections_status, 'insufficient_data') AS connections_status,
    c.connections_utilization_pct,
    c.connections_headroom,
    CASE
        WHEN ms.status IS NULL AND mw.status IS NULL THEN 'insufficient_data'
        WHEN ms.status = 'critical' OR mw.status = 'critical' THEN 'critical'
        WHEN ms.status = 'warning' OR mw.status = 'warning' THEN 'warning'
        ELSE COALESCE(ms.status, mw.status, 'healthy')
    END AS memory_status,
    GREATEST(0, LEAST(100, round(
        COALESCE(ms.utilization_pct, 0) * 0.6 +
        COALESCE(mw.utilization_pct, 0) * 0.4,
        1
    ))) AS memory_pressure_score,
    COALESCE(io.io_status, 'insufficient_data') AS io_status,
    io.io_saturation_pct,
    COALESCE(s.storage_status, 'insufficient_data') AS storage_status,
    s.storage_utilization_pct,
    CASE
        WHEN s.storage_recommendation ~ 'growing [0-9\.]+' THEN
            (regexp_match(s.storage_recommendation, 'growing ([0-9\.]+)'))[1]::numeric
        ELSE NULL
    END AS storage_growth_mb_per_day,
    CASE
        WHEN 'critical' IN (
            c.connections_status,
            CASE WHEN ms.status = 'critical' OR mw.status = 'critical' THEN 'critical' END,
            io.io_status,
            s.storage_status
        ) THEN 'critical'
        WHEN 'warning' IN (
            c.connections_status,
            CASE WHEN ms.status = 'warning' OR mw.status = 'warning' THEN 'warning' END,
            io.io_status,
            s.storage_status
        ) THEN 'warning'
        WHEN 'insufficient_data' IN (
            COALESCE(c.connections_status, 'insufficient_data'),
            CASE WHEN ms.status IS NULL AND mw.status IS NULL THEN 'insufficient_data' END,
            COALESCE(io.io_status, 'insufficient_data'),
            COALESCE(s.storage_status, 'insufficient_data')
        ) THEN 'insufficient_data'
        ELSE 'healthy'
    END AS overall_status,
    ARRAY_REMOVE(ARRAY[
        CASE WHEN c.connections_status IN ('critical', 'warning')
             THEN 'CONNECTIONS: ' || c.connections_recommendation END,
        CASE WHEN ms.status IN ('critical', 'warning')
             THEN 'MEMORY (shared_buffers): ' ||
                  (SELECT recommendation FROM capacity_metrics WHERE metric = 'memory_shared_buffers') END,
        CASE WHEN mw.status IN ('critical', 'warning')
             THEN 'MEMORY (work_mem): ' ||
                  (SELECT recommendation FROM capacity_metrics WHERE metric = 'memory_work_mem') END,
        CASE WHEN io.io_status IN ('critical', 'warning')
             THEN 'I/O: ' || io.io_recommendation END,
        CASE WHEN s.storage_status IN ('critical', 'warning')
             THEN 'STORAGE: ' || s.storage_recommendation END
    ], NULL) AS critical_issues
FROM latest_snapshot l
LEFT JOIN connections_metric c ON true
LEFT JOIN memory_sb_metric ms ON true
LEFT JOIN memory_wm_metric mw ON true
LEFT JOIN io_metric io ON true
LEFT JOIN storage_metric s ON true;
COMMENT ON VIEW pgfr_analyze.capacity_dashboard IS
'At-a-glance capacity planning dashboard. Shows current status (healthy/warning/critical) across all resource dimensions: connections, memory, I/O, storage. Includes utilization percentages, composite memory pressure score, and array of critical issues requiring attention. Based on last 24 hours of data. Part of Phase 1 MVP capacity planning enhancements (FR-3.1).';

-- NOTE: Storm and Regression dashboard views removed in 2.21
-- Use detect_query_storms() and detect_regressions() for on-demand analysis

-- TABLE-LEVEL HOTSPOT TRACKING ANALYSIS FUNCTIONS
-- =============================================================================

-- Compares table activity between two time points
-- Returns delta metrics for DML activity, scans, and maintenance events

-- ==========================================================
-- pgfr_analyze/sql/06_table_analysis.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

CREATE OR REPLACE FUNCTION pgfr_analyze.table_compare(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ,
    p_limit INTEGER DEFAULT 25
)
RETURNS TABLE(
    schemaname              TEXT,
    relname                 TEXT,
    relid                   OID,
    seq_scan_delta          BIGINT,
    seq_tup_read_delta      BIGINT,
    idx_scan_delta          BIGINT,
    idx_tup_fetch_delta     BIGINT,
    n_tup_ins_delta         BIGINT,
    n_tup_upd_delta         BIGINT,
    n_tup_del_delta         BIGINT,
    n_tup_hot_upd_delta     BIGINT,
    dead_tup_pct            NUMERIC,
    vacuum_count_delta      BIGINT,
    autovacuum_count_delta  BIGINT,
    analyze_count_delta     BIGINT,
    autoanalyze_count_delta BIGINT,
    total_activity          BIGINT
)
LANGUAGE sql STABLE AS $$
    WITH
    start_snap AS (
        SELECT DISTINCT ON (ts.relid) ts.*
        FROM pgfr_record.table_snapshots ts
        JOIN pgfr_record.snapshots s ON s.id = ts.snapshot_id
        WHERE s.captured_at <= p_start_time
        ORDER BY ts.relid, s.captured_at DESC
    ),
    end_snap AS (
        SELECT DISTINCT ON (ts.relid) ts.*
        FROM pgfr_record.table_snapshots ts
        JOIN pgfr_record.snapshots s ON s.id = ts.snapshot_id
        WHERE s.captured_at >= p_end_time
        ORDER BY ts.relid, s.captured_at ASC
    ),
    matched AS (
        SELECT
            COALESCE(e.schemaname, split_part(e.relid::regclass::text, '.', 1)) AS schemaname,
            COALESCE(e.relname, split_part(e.relid::regclass::text, '.', 2)) AS relname,
            e.relid,
            COALESCE(e.seq_scan, 0) - COALESCE(s.seq_scan, 0) AS seq_scan_delta,
            COALESCE(e.seq_tup_read, 0) - COALESCE(s.seq_tup_read, 0) AS seq_tup_read_delta,
            COALESCE(e.idx_scan, 0) - COALESCE(s.idx_scan, 0) AS idx_scan_delta,
            COALESCE(e.idx_tup_fetch, 0) - COALESCE(s.idx_tup_fetch, 0) AS idx_tup_fetch_delta,
            COALESCE(e.n_tup_ins, 0) - COALESCE(s.n_tup_ins, 0) AS n_tup_ins_delta,
            COALESCE(e.n_tup_upd, 0) - COALESCE(s.n_tup_upd, 0) AS n_tup_upd_delta,
            COALESCE(e.n_tup_del, 0) - COALESCE(s.n_tup_del, 0) AS n_tup_del_delta,
            COALESCE(e.n_tup_hot_upd, 0) - COALESCE(s.n_tup_hot_upd, 0) AS n_tup_hot_upd_delta,
            e.n_live_tup,
            e.n_dead_tup,
            COALESCE(e.vacuum_count, 0) - COALESCE(s.vacuum_count, 0) AS vacuum_count_delta,
            COALESCE(e.autovacuum_count, 0) - COALESCE(s.autovacuum_count, 0) AS autovacuum_count_delta,
            COALESCE(e.analyze_count, 0) - COALESCE(s.analyze_count, 0) AS analyze_count_delta,
            COALESCE(e.autoanalyze_count, 0) - COALESCE(s.autoanalyze_count, 0) AS autoanalyze_count_delta
        FROM end_snap e
        LEFT JOIN start_snap s ON s.relid = e.relid
    )
    SELECT
        m.schemaname,
        m.relname,
        m.relid,
        m.seq_scan_delta,
        m.seq_tup_read_delta,
        m.idx_scan_delta,
        m.idx_tup_fetch_delta,
        m.n_tup_ins_delta,
        m.n_tup_upd_delta,
        m.n_tup_del_delta,
        m.n_tup_hot_upd_delta,
        CASE
            WHEN COALESCE(m.n_live_tup, 0) > 0
            THEN round(100.0 * COALESCE(m.n_dead_tup, 0) / (COALESCE(m.n_live_tup, 0) + COALESCE(m.n_dead_tup, 0)), 1)
            ELSE 0
        END AS dead_tup_pct,
        m.vacuum_count_delta,
        m.autovacuum_count_delta,
        m.analyze_count_delta,
        m.autoanalyze_count_delta,
        (m.seq_tup_read_delta + m.idx_tup_fetch_delta +
         m.n_tup_ins_delta + m.n_tup_upd_delta + m.n_tup_del_delta) AS total_activity
    FROM matched m
    WHERE (m.seq_tup_read_delta + m.idx_tup_fetch_delta +
           m.n_tup_ins_delta + m.n_tup_upd_delta + m.n_tup_del_delta) > 0
    ORDER BY total_activity DESC
    LIMIT p_limit
$$;
COMMENT ON FUNCTION pgfr_analyze.table_compare(TIMESTAMPTZ, TIMESTAMPTZ, INTEGER) IS
'Compare table activity between two time points. Shows DML deltas, scan counts, dead tuple percentage, and maintenance events. Useful for identifying hot tables during incidents.';


-- Identifies table hotspots and potential issues
-- Returns actionable recommendations for tables with problems
CREATE OR REPLACE FUNCTION pgfr_analyze.table_hotspots(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TABLE(
    schemaname      TEXT,
    relname         TEXT,
    issue_type      TEXT,
    severity        TEXT,
    description     TEXT,
    recommendation  TEXT
)
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_table RECORD;
    v_hot_ratio NUMERIC;
BEGIN
    FOR v_table IN
        SELECT * FROM pgfr_analyze.table_compare(p_start_time, p_end_time, 100)
    LOOP
        -- High sequential scan activity
        IF v_table.seq_scan_delta > 100 AND v_table.seq_tup_read_delta > 100000 THEN
            schemaname := v_table.schemaname;
            relname := v_table.relname;
            issue_type := 'SEQUENTIAL_SCAN_STORM';
            severity := CASE
                WHEN v_table.seq_tup_read_delta > 10000000 THEN 'high'
                WHEN v_table.seq_tup_read_delta > 1000000 THEN 'medium'
                ELSE 'low'
            END;
            description := format('%s sequential scans reading %s tuples',
                                 v_table.seq_scan_delta,
                                 v_table.seq_tup_read_delta);
            recommendation := 'Consider adding an index or reviewing query WHERE clauses';
            RETURN NEXT;
        END IF;

        -- High dead tuple percentage (bloat)
        IF v_table.dead_tup_pct > 20 THEN
            schemaname := v_table.schemaname;
            relname := v_table.relname;
            issue_type := 'TABLE_BLOAT';
            severity := CASE
                WHEN v_table.dead_tup_pct > 50 THEN 'high'
                WHEN v_table.dead_tup_pct > 30 THEN 'medium'
                ELSE 'low'
            END;
            description := format('%s%% dead tuples', round(v_table.dead_tup_pct));
            recommendation := 'Run VACUUM or check autovacuum settings';
            RETURN NEXT;
        END IF;

        -- Low HOT update ratio (inefficient updates)
        IF v_table.n_tup_upd_delta > 1000 THEN
            v_hot_ratio := CASE
                WHEN v_table.n_tup_upd_delta > 0
                THEN 100.0 * v_table.n_tup_hot_upd_delta / v_table.n_tup_upd_delta
                ELSE 100
            END;

            IF v_hot_ratio < 50 THEN
                schemaname := v_table.schemaname;
                relname := v_table.relname;
                issue_type := 'LOW_HOT_UPDATE_RATIO';
                severity := 'medium';
                description := format('%s updates, only %s%% HOT',
                                     v_table.n_tup_upd_delta,
                                     round(v_hot_ratio, 1));
                recommendation := 'Consider increasing fillfactor or reducing indexed columns';
                RETURN NEXT;
            END IF;
        END IF;

        -- Frequent autovacuum (indicates high churn)
        IF v_table.autovacuum_count_delta > 5 THEN
            schemaname := v_table.schemaname;
            relname := v_table.relname;
            issue_type := 'HIGH_AUTOVACUUM_FREQUENCY';
            severity := 'low';
            description := format('%s autovacuums during period',
                                 v_table.autovacuum_count_delta);
            recommendation := 'High write activity detected; ensure autovacuum keeps up';
            RETURN NEXT;
        END IF;
    END LOOP;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.table_hotspots(TIMESTAMPTZ, TIMESTAMPTZ) IS
'Identify table-level hotspots and issues. Returns actionable recommendations for sequential scan storms, table bloat, low HOT update ratios, and frequent autovacuum activity.';


-- =============================================================================
-- INDEX USAGE TRACKING ANALYSIS FUNCTIONS
-- =============================================================================

-- Identifies unused or rarely used indexes
-- Returns indexes that may be candidates for removal

-- ==========================================================
-- pgfr_analyze/sql/07_index_analysis.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

CREATE OR REPLACE FUNCTION pgfr_analyze.unused_indexes(
    p_lookback_interval INTERVAL DEFAULT '7 days'
)
RETURNS TABLE(
    schemaname      TEXT,
    relname         TEXT,
    indexrelname    TEXT,
    index_size      TEXT,
    last_scan_count BIGINT,
    recommendation  TEXT
)
LANGUAGE sql STABLE AS $$
    WITH latest_snapshot AS (
        SELECT max(id) AS snapshot_id
        FROM pgfr_record.snapshots
        WHERE captured_at > now() - p_lookback_interval
    ),
    earliest_snapshot AS (
        SELECT min(id) AS snapshot_id
        FROM pgfr_record.snapshots
        WHERE captured_at > now() - p_lookback_interval
    ),
    index_usage AS (
        SELECT
            COALESCE(e.schemaname, split_part(e.relid::regclass::text, '.', 1)) AS schemaname,
            COALESCE(e.relname, split_part(e.relid::regclass::text, '.', 2)) AS relname,
            COALESCE(e.indexrelname, split_part(e.indexrelid::regclass::text, '.', 2)) AS indexrelname,
            e.indexrelid,
            e.index_size_bytes,
            COALESCE(e.idx_scan, 0) - COALESCE(s.idx_scan, 0) AS scan_delta
        FROM pgfr_record.index_snapshots e
        CROSS JOIN latest_snapshot ls
        LEFT JOIN pgfr_record.index_snapshots s
            ON s.indexrelid = e.indexrelid
            AND s.snapshot_id = (SELECT snapshot_id FROM earliest_snapshot)
        WHERE e.snapshot_id = ls.snapshot_id
    )
    SELECT
        iu.schemaname,
        iu.relname,
        iu.indexrelname,
        pgfr_record._pretty_bytes(iu.index_size_bytes) AS index_size,
        iu.scan_delta AS last_scan_count,
        CASE
            WHEN iu.scan_delta = 0 THEN 'DROP INDEX (never used in ' || p_lookback_interval::text || ')'
            WHEN iu.scan_delta < 10 THEN 'Consider dropping (rarely used)'
            ELSE 'Keep (actively used)'
        END AS recommendation
    FROM index_usage iu
    WHERE iu.scan_delta < 100  -- Threshold for "rarely used"
        AND iu.indexrelname NOT LIKE '%_pkey'  -- Don't suggest dropping primary keys
    ORDER BY iu.index_size_bytes DESC
$$;
COMMENT ON FUNCTION pgfr_analyze.unused_indexes(INTERVAL) IS
'Identify unused or rarely used indexes. Returns indexes that may be candidates for removal to save space and improve write performance. Default lookback is 7 days.';


-- Analyzes index efficiency and usage patterns
-- Returns selectivity and scans-per-GB metrics
CREATE OR REPLACE FUNCTION pgfr_analyze.index_efficiency(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ,
    p_limit INTEGER DEFAULT 25
)
RETURNS TABLE(
    schemaname          TEXT,
    relname             TEXT,
    indexrelname        TEXT,
    idx_scan_delta      BIGINT,
    idx_tup_read_delta  BIGINT,
    idx_tup_fetch_delta BIGINT,
    selectivity         NUMERIC,
    index_size          TEXT,
    scans_per_gb        NUMERIC
)
LANGUAGE sql STABLE AS $$
    WITH
    start_snap AS (
        SELECT DISTINCT ON (i.indexrelid) i.*
        FROM pgfr_record.index_snapshots i
        JOIN pgfr_record.snapshots s ON s.id = i.snapshot_id
        WHERE s.captured_at <= p_start_time
        ORDER BY i.indexrelid, s.captured_at DESC
    ),
    end_snap AS (
        SELECT DISTINCT ON (i.indexrelid) i.*
        FROM pgfr_record.index_snapshots i
        JOIN pgfr_record.snapshots s ON s.id = i.snapshot_id
        WHERE s.captured_at >= p_end_time
        ORDER BY i.indexrelid, s.captured_at ASC
    )
    SELECT
        COALESCE(e.schemaname, split_part(e.relid::regclass::text, '.', 1)) AS schemaname,
        COALESCE(e.relname, split_part(e.relid::regclass::text, '.', 2)) AS relname,
        COALESCE(e.indexrelname, split_part(e.indexrelid::regclass::text, '.', 2)) AS indexrelname,
        COALESCE(e.idx_scan, 0) - COALESCE(s.idx_scan, 0) AS idx_scan_delta,
        COALESCE(e.idx_tup_read, 0) - COALESCE(s.idx_tup_read, 0) AS idx_tup_read_delta,
        COALESCE(e.idx_tup_fetch, 0) - COALESCE(s.idx_tup_fetch, 0) AS idx_tup_fetch_delta,
        CASE
            WHEN (COALESCE(e.idx_tup_read, 0) - COALESCE(s.idx_tup_read, 0)) > 0
            THEN round(100.0 * (COALESCE(e.idx_tup_fetch, 0) - COALESCE(s.idx_tup_fetch, 0)) /
                             (COALESCE(e.idx_tup_read, 0) - COALESCE(s.idx_tup_read, 0)), 1)
            ELSE NULL
        END AS selectivity,
        pgfr_record._pretty_bytes(e.index_size_bytes) AS index_size,
        CASE
            WHEN COALESCE(e.index_size_bytes, 0) > 0
            THEN round((COALESCE(e.idx_scan, 0) - COALESCE(s.idx_scan, 0)) /
                      (e.index_size_bytes / 1073741824.0::numeric), 2)
            ELSE NULL
        END AS scans_per_gb
    FROM end_snap e
    LEFT JOIN start_snap s ON s.indexrelid = e.indexrelid
    WHERE (COALESCE(e.idx_scan, 0) - COALESCE(s.idx_scan, 0)) > 0
    ORDER BY idx_scan_delta DESC
    LIMIT p_limit
$$;
COMMENT ON FUNCTION pgfr_analyze.index_efficiency(TIMESTAMPTZ, TIMESTAMPTZ, INTEGER) IS
'Analyze index efficiency and usage patterns. Returns selectivity (fetch/read ratio) and scans-per-GB metrics. Low selectivity may indicate poor index choices.';


-- =============================================================================
-- CONFIGURATION SNAPSHOT ANALYSIS FUNCTIONS
-- =============================================================================

-- Detects configuration changes between two time points
-- Returns parameters that changed with old and new values

-- ==========================================================
-- pgfr_analyze/sql/08_config.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

CREATE OR REPLACE FUNCTION pgfr_analyze.config_changes(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TABLE(
    parameter_name  TEXT,
    old_value       TEXT,
    new_value       TEXT,
    old_source      TEXT,
    new_source      TEXT,
    changed_at      TIMESTAMPTZ
)
LANGUAGE sql STABLE AS $$
    WITH
    start_configs AS (
        SELECT DISTINCT ON (cs.name) cs.name, cs.setting, cs.unit, cs.source, s.captured_at
        FROM pgfr_record.config_snapshots cs
        JOIN pgfr_record.snapshots s ON s.id = cs.snapshot_id
        WHERE s.captured_at <= p_start_time
        ORDER BY cs.name, s.captured_at DESC
    ),
    end_configs AS (
        SELECT DISTINCT ON (cs.name) cs.name, cs.setting, cs.unit, cs.source, s.captured_at
        FROM pgfr_record.config_snapshots cs
        JOIN pgfr_record.snapshots s ON s.id = cs.snapshot_id
        WHERE s.captured_at >= p_end_time
        ORDER BY cs.name, s.captured_at ASC
    )
    SELECT
        COALESCE(e.name, s.name) AS parameter_name,
        s.setting || COALESCE(' ' || s.unit, '') AS old_value,
        e.setting || COALESCE(' ' || e.unit, '') AS new_value,
        s.source AS old_source,
        e.source AS new_source,
        e.captured_at AS changed_at
    FROM end_configs e
    FULL OUTER JOIN start_configs s ON s.name = e.name
    WHERE e.setting IS DISTINCT FROM s.setting
        OR e.source IS DISTINCT FROM s.source
    ORDER BY parameter_name
$$;
COMMENT ON FUNCTION pgfr_analyze.config_changes(TIMESTAMPTZ, TIMESTAMPTZ) IS
'Detect PostgreSQL configuration changes between two time points. Useful for correlating configuration changes with performance incidents.';


-- Retrieves configuration at a specific point in time
-- Optionally filters by parameter name prefix (category)
CREATE OR REPLACE FUNCTION pgfr_analyze.config_at(
    p_timestamp TIMESTAMPTZ,
    p_category TEXT DEFAULT NULL
)
RETURNS TABLE(
    parameter_name  TEXT,
    value           TEXT,
    source          TEXT
)
LANGUAGE sql STABLE AS $$
    SELECT DISTINCT ON (cs.name)
        cs.name AS parameter_name,
        cs.setting || COALESCE(' ' || cs.unit, '') AS value,
        cs.source
    FROM pgfr_record.config_snapshots cs
    JOIN pgfr_record.snapshots s ON s.id = cs.snapshot_id
    WHERE s.captured_at <= p_timestamp
        AND (p_category IS NULL OR cs.name LIKE p_category || '%')
    ORDER BY cs.name, s.captured_at DESC
$$;
COMMENT ON FUNCTION pgfr_analyze.config_at(TIMESTAMPTZ, TEXT) IS
'Retrieve PostgreSQL configuration at a specific point in time. Optionally filter by category prefix (e.g., ''autovacuum'', ''work_mem'').';


-- Performs a health check on current PostgreSQL configuration
-- Returns potential issues and recommendations
CREATE OR REPLACE FUNCTION pgfr_analyze.config_health_check()
RETURNS TABLE(
    category        TEXT,
    parameter_name  TEXT,
    current_value   TEXT,
    issue           TEXT,
    recommendation  TEXT
)
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_shared_buffers BIGINT;
    v_work_mem BIGINT;
    v_max_connections INTEGER;
BEGIN
    -- Get current values
    SELECT setting::bigint * 8192 INTO v_shared_buffers
    FROM pg_settings WHERE name = 'shared_buffers';

    SELECT setting::bigint * 1024 INTO v_work_mem
    FROM pg_settings WHERE name = 'work_mem';

    SELECT setting::integer INTO v_max_connections
    FROM pg_settings WHERE name = 'max_connections';

    -- Check shared_buffers (should be at least 128 MB for most workloads)
    IF v_shared_buffers < 134217728 THEN  -- < 128 MB
        category := 'memory';
        parameter_name := 'shared_buffers';
        current_value := pgfr_record._pretty_bytes(v_shared_buffers);
        issue := 'Very low shared_buffers';
        recommendation := 'Increase to at least 25% of available RAM';
        RETURN NEXT;
    END IF;

    -- Check work_mem (should be at least 16MB for analytical workloads)
    IF v_work_mem < 16777216 THEN  -- < 16 MB
        category := 'memory';
        parameter_name := 'work_mem';
        current_value := pgfr_record._pretty_bytes(v_work_mem);
        issue := 'Low work_mem may cause disk spills';
        recommendation := 'Consider increasing to 32-64MB, depending on workload';
        RETURN NEXT;
    END IF;

    -- Check max_connections (high values waste RAM)
    IF v_max_connections > 200 THEN
        category := 'connections';
        parameter_name := 'max_connections';
        current_value := v_max_connections::text;
        issue := 'High max_connections wastes memory';
        recommendation := 'Use connection pooling (pgBouncer) instead of high max_connections';
        RETURN NEXT;
    END IF;

    -- Check if statement timeout is set
    IF NOT EXISTS (
        SELECT 1 FROM pg_settings
        WHERE name = 'statement_timeout' AND setting != '0'
    ) THEN
        category := 'safety';
        parameter_name := 'statement_timeout';
        current_value := 'disabled';
        issue := 'No statement timeout protection';
        recommendation := 'Set statement_timeout to prevent runaway queries (e.g., 30s-5min)';
        RETURN NEXT;
    END IF;

    RETURN;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.config_health_check() IS
'Perform a health check on current PostgreSQL configuration. Returns potential issues and recommendations for memory, connections, and safety settings.';


-- =============================================================================
-- DATABASE/ROLE CONFIGURATION ANALYSIS FUNCTIONS
-- =============================================================================

-- Retrieves database/role configuration overrides at a specific point in time
-- Optionally filters by database, role, or parameter name prefix
CREATE OR REPLACE FUNCTION pgfr_analyze.db_role_config_at(
    p_timestamp TIMESTAMPTZ,
    p_database TEXT DEFAULT NULL,
    p_role TEXT DEFAULT NULL,
    p_prefix TEXT DEFAULT NULL
)
RETURNS TABLE(
    database_name   TEXT,
    role_name       TEXT,
    parameter_name  TEXT,
    parameter_value TEXT,
    scope           TEXT
)
LANGUAGE sql STABLE AS $$
    SELECT DISTINCT ON (drc.database_name, drc.role_name, drc.parameter_name)
        NULLIF(drc.database_name, '') AS database_name,
        NULLIF(drc.role_name, '') AS role_name,
        drc.parameter_name,
        drc.parameter_value,
        CASE
            WHEN drc.database_name <> '' AND drc.role_name <> '' THEN 'database+role'
            WHEN drc.database_name <> '' THEN 'database'
            WHEN drc.role_name <> '' THEN 'role'
            ELSE 'unknown'
        END AS scope
    FROM pgfr_record.db_role_config_snapshots drc
    JOIN pgfr_record.snapshots s ON s.id = drc.snapshot_id
    WHERE s.captured_at <= p_timestamp
        AND (p_database IS NULL OR drc.database_name = p_database)
        AND (p_role IS NULL OR drc.role_name = p_role)
        AND (p_prefix IS NULL OR drc.parameter_name LIKE p_prefix || '%')
    ORDER BY drc.database_name, drc.role_name, drc.parameter_name, s.captured_at DESC
$$;
COMMENT ON FUNCTION pgfr_analyze.db_role_config_at(TIMESTAMPTZ, TEXT, TEXT, TEXT) IS
'Retrieve database/role configuration overrides at a specific point in time. Filter by database, role, or parameter prefix.';


-- Detects database/role configuration changes between two time points
-- Returns parameters that were added, removed, or modified
CREATE OR REPLACE FUNCTION pgfr_analyze.db_role_config_changes(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TABLE(
    database_name   TEXT,
    role_name       TEXT,
    parameter_name  TEXT,
    old_value       TEXT,
    new_value       TEXT,
    change_type     TEXT
)
LANGUAGE sql STABLE AS $$
    WITH
    start_configs AS (
        SELECT DISTINCT ON (drc.database_name, drc.role_name, drc.parameter_name)
            drc.database_name, drc.role_name, drc.parameter_name, drc.parameter_value
        FROM pgfr_record.db_role_config_snapshots drc
        JOIN pgfr_record.snapshots s ON s.id = drc.snapshot_id
        WHERE s.captured_at <= p_start_time
        ORDER BY drc.database_name, drc.role_name, drc.parameter_name, s.captured_at DESC
    ),
    end_configs AS (
        SELECT DISTINCT ON (drc.database_name, drc.role_name, drc.parameter_name)
            drc.database_name, drc.role_name, drc.parameter_name, drc.parameter_value
        FROM pgfr_record.db_role_config_snapshots drc
        JOIN pgfr_record.snapshots s ON s.id = drc.snapshot_id
        WHERE s.captured_at <= p_end_time
        ORDER BY drc.database_name, drc.role_name, drc.parameter_name, s.captured_at DESC
    )
    SELECT
        NULLIF(COALESCE(e.database_name, s.database_name), '') AS database_name,
        NULLIF(COALESCE(e.role_name, s.role_name), '') AS role_name,
        COALESCE(e.parameter_name, s.parameter_name) AS parameter_name,
        s.parameter_value AS old_value,
        e.parameter_value AS new_value,
        CASE
            WHEN s.parameter_name IS NULL THEN 'added'
            WHEN e.parameter_name IS NULL THEN 'removed'
            ELSE 'modified'
        END AS change_type
    FROM end_configs e
    FULL OUTER JOIN start_configs s
        ON s.database_name = e.database_name
        AND s.role_name = e.role_name
        AND s.parameter_name = e.parameter_name
    WHERE e.parameter_value IS DISTINCT FROM s.parameter_value
    ORDER BY database_name NULLS FIRST, role_name NULLS FIRST, parameter_name
$$;
COMMENT ON FUNCTION pgfr_analyze.db_role_config_changes(TIMESTAMPTZ, TIMESTAMPTZ) IS
'Detect database/role configuration changes between two time points. Returns added, removed, and modified settings.';


-- Provides a summary overview of all database/role configuration overrides
-- Groups by scope (database-only, role-only, or database+role combination)
CREATE OR REPLACE FUNCTION pgfr_analyze.db_role_config_summary()
RETURNS TABLE(
    scope           TEXT,
    database_name   TEXT,
    role_name       TEXT,
    parameter_count BIGINT,
    parameters      TEXT[]
)
LANGUAGE sql STABLE AS $$
    WITH latest_snapshot AS (
        SELECT id FROM pgfr_record.snapshots ORDER BY captured_at DESC LIMIT 1
    ),
    config_data AS (
        SELECT
            NULLIF(drc.database_name, '') AS database_name,
            NULLIF(drc.role_name, '') AS role_name,
            drc.parameter_name,
            CASE
                WHEN drc.database_name <> '' AND drc.role_name <> '' THEN 'database+role'
                WHEN drc.database_name <> '' THEN 'database'
                WHEN drc.role_name <> '' THEN 'role'
                ELSE 'unknown'
            END AS scope
        FROM pgfr_record.db_role_config_snapshots drc
        WHERE drc.snapshot_id = (SELECT id FROM latest_snapshot)
    )
    SELECT
        scope,
        database_name,
        role_name,
        count(*) AS parameter_count,
        array_agg(parameter_name ORDER BY parameter_name) AS parameters
    FROM config_data
    GROUP BY scope, database_name, role_name
    ORDER BY scope, database_name NULLS FIRST, role_name NULLS FIRST
$$;
COMMENT ON FUNCTION pgfr_analyze.db_role_config_summary() IS
'Overview of database/role configuration overrides grouped by scope. Shows which databases and roles have custom settings.';


-- =============================================================================
-- TIME-TRAVEL DEBUGGING
-- =============================================================================
-- Enables forensic analysis of "what happened at exactly 10:23:47?"
-- Bridges the gap between sample intervals by interpolating system metrics
-- and surfacing exact-timestamp events from activity samples


-- Main time-travel analysis function
-- Provides interpolated system state at any arbitrary timestamp
-- Input: Target timestamp, context window (default 5 minutes)
-- Output: Interpolated metrics, events, sessions, locks, wait events, confidence, recommendations

-- ==========================================================
-- pgfr_analyze/sql/09_incident_analysis.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

CREATE OR REPLACE FUNCTION pgfr_analyze.what_happened_at(
    p_timestamp TIMESTAMPTZ,
    p_context_window INTERVAL DEFAULT '5 minutes'
)
RETURNS TABLE(
    -- Temporal context
    requested_time TIMESTAMPTZ,
    sample_before TIMESTAMPTZ,
    sample_after TIMESTAMPTZ,
    snapshot_before TIMESTAMPTZ,
    snapshot_after TIMESTAMPTZ,

    -- Interpolated state from snapshots
    est_connections_active NUMERIC,
    est_connections_total NUMERIC,
    est_xact_rate NUMERIC,
    est_blks_hit_ratio NUMERIC,

    -- Exact-timestamp events (from activity samples)
    events JSONB,

    -- Activity analysis (from activity samples ring buffer)
    sessions_active INTEGER,
    long_running_queries INTEGER,
    longest_query_secs NUMERIC,

    -- Lock analysis
    lock_contention_detected BOOLEAN,
    blocked_sessions INTEGER,

    -- Wait events
    top_wait_events JSONB,

    -- Confidence assessment
    confidence TEXT,
    confidence_score NUMERIC,
    data_quality_notes TEXT[],

    -- Actionable recommendations
    recommendations TEXT[]
)
LANGUAGE plpgsql AS $$
DECLARE
    -- Snapshot data
    v_snap_before RECORD;
    v_snap_after RECORD;
    v_snap_gap_secs NUMERIC;

    -- Sample data
    v_sample_before RECORD;
    v_sample_after RECORD;
    v_sample_gap_secs NUMERIC;

    -- Interpolated values
    v_est_active NUMERIC;
    v_est_total NUMERIC;
    v_est_xact_rate NUMERIC;
    v_est_hit_ratio NUMERIC;

    -- Events array
    v_events JSONB := '[]'::jsonb;
    v_event JSONB;

    -- Activity analysis
    v_sessions INTEGER := 0;
    v_long_running INTEGER := 0;
    v_longest_secs NUMERIC := 0;

    -- Lock analysis
    v_lock_detected BOOLEAN := FALSE;
    v_blocked INTEGER := 0;

    -- Wait events
    v_waits JSONB := '[]'::jsonb;

    -- Confidence calculation
    v_confidence_score NUMERIC := 0.5;
    v_confidence_level TEXT := 'low';
    v_notes TEXT[] := ARRAY[]::TEXT[];
    v_recs TEXT[] := ARRAY[]::TEXT[];

    -- Context window bounds
    v_window_start TIMESTAMPTZ;
    v_window_end TIMESTAMPTZ;

    -- v2 ring fallback (used when legacy samples_ring is empty)
    v_v2_sample_ts_before int4;
    v_v2_sample_ts_after  int4;
    v_v2_captured_before  timestamptz;
    v_v2_captured_after   timestamptz;
BEGIN
    -- Calculate window bounds
    v_window_start := p_timestamp - p_context_window;
    v_window_end := p_timestamp + p_context_window;

    -- ==========================================================================
    -- STEP 1: Find surrounding snapshots
    -- ==========================================================================
    SELECT * INTO v_snap_before
    FROM pgfr_record.snapshots
    WHERE captured_at <= p_timestamp
    ORDER BY captured_at DESC
    LIMIT 1;

    SELECT * INTO v_snap_after
    FROM pgfr_record.snapshots
    WHERE captured_at >= p_timestamp
    ORDER BY captured_at ASC
    LIMIT 1;

    -- Calculate snapshot gap
    IF v_snap_before IS NOT NULL AND v_snap_after IS NOT NULL THEN
        v_snap_gap_secs := EXTRACT(EPOCH FROM (v_snap_after.captured_at - v_snap_before.captured_at));
    END IF;

    -- ==========================================================================
    -- STEP 2: Find surrounding samples (from ring buffer)
    -- ==========================================================================
    SELECT sr.* INTO v_sample_before
    FROM pgfr_record.samples_ring sr
    WHERE sr.captured_at <= p_timestamp
      AND sr.captured_at > '1970-01-01'::timestamptz
    ORDER BY sr.captured_at DESC
    LIMIT 1;

    SELECT sr.* INTO v_sample_after
    FROM pgfr_record.samples_ring sr
    WHERE sr.captured_at >= p_timestamp
      AND sr.captured_at > '1970-01-01'::timestamptz
    ORDER BY sr.captured_at ASC
    LIMIT 1;

    -- Calculate sample gap
    IF v_sample_before IS NOT NULL AND v_sample_after IS NOT NULL THEN
        v_sample_gap_secs := EXTRACT(EPOCH FROM (v_sample_after.captured_at - v_sample_before.captured_at));
    END IF;

    -- v2 ring fallback: if legacy samples_ring is empty, use activity_samples
    IF v_sample_before IS NULL THEN
        SELECT sample_ts,
               pgfr_record.epoch() + sample_ts * interval '1 second'
        INTO   v_v2_sample_ts_before, v_v2_captured_before
        FROM   pgfr_record.activity_samples
        WHERE  pgfr_record.epoch() + sample_ts * interval '1 second' <= p_timestamp
        ORDER BY sample_ts DESC
        LIMIT 1;
    END IF;
    IF v_sample_after IS NULL THEN
        SELECT sample_ts,
               pgfr_record.epoch() + sample_ts * interval '1 second'
        INTO   v_v2_sample_ts_after, v_v2_captured_after
        FROM   pgfr_record.activity_samples
        WHERE  pgfr_record.epoch() + sample_ts * interval '1 second' >= p_timestamp
        ORDER BY sample_ts ASC
        LIMIT 1;
    END IF;
    IF v_v2_sample_ts_before IS NOT NULL AND v_v2_sample_ts_after IS NOT NULL AND v_sample_gap_secs IS NULL THEN
        v_sample_gap_secs := v_v2_sample_ts_after - v_v2_sample_ts_before;
    END IF;

    -- ==========================================================================
    -- STEP 3: Interpolate snapshot metrics
    -- ==========================================================================
    IF v_snap_before IS NOT NULL AND v_snap_after IS NOT NULL THEN
        v_est_active := pgfr_record._interpolate_metric(
            v_snap_before.connections_active::NUMERIC,
            v_snap_before.captured_at,
            v_snap_after.connections_active::NUMERIC,
            v_snap_after.captured_at,
            p_timestamp
        );

        v_est_total := pgfr_record._interpolate_metric(
            v_snap_before.connections_total::NUMERIC,
            v_snap_before.captured_at,
            v_snap_after.connections_total::NUMERIC,
            v_snap_after.captured_at,
            p_timestamp
        );

        -- Calculate transaction rate delta
        IF v_snap_gap_secs > 0 AND
           v_snap_before.xact_commit IS NOT NULL AND
           v_snap_after.xact_commit IS NOT NULL THEN
            v_est_xact_rate := round(
                ((v_snap_after.xact_commit - v_snap_before.xact_commit) +
                 (v_snap_after.xact_rollback - v_snap_before.xact_rollback))::NUMERIC / v_snap_gap_secs,
                1
            );
        END IF;

        -- Calculate buffer hit ratio
        IF v_snap_after.blks_hit IS NOT NULL AND
           v_snap_after.blks_read IS NOT NULL AND
           (v_snap_after.blks_hit + v_snap_after.blks_read) > 0 THEN
            v_est_hit_ratio := round(
                v_snap_after.blks_hit::NUMERIC /
                NULLIF(v_snap_after.blks_hit + v_snap_after.blks_read, 0) * 100,
                2
            );
        END IF;
    ELSIF v_snap_before IS NOT NULL THEN
        -- Only have before snapshot - use its values
        v_est_active := v_snap_before.connections_active;
        v_est_total := v_snap_before.connections_total;
        v_notes := array_append(v_notes, 'No snapshot after target time - using last known values');
    ELSIF v_snap_after IS NOT NULL THEN
        -- Only have after snapshot - use its values
        v_est_active := v_snap_after.connections_active;
        v_est_total := v_snap_after.connections_total;
        v_notes := array_append(v_notes, 'No snapshot before target time - using next known values');
    ELSE
        v_notes := array_append(v_notes, 'No snapshots found in range');
    END IF;

    -- ==========================================================================
    -- STEP 4: Collect exact-timestamp events from activity samples
    -- ==========================================================================

    -- Check for checkpoint event near target time
    IF v_snap_before IS NOT NULL AND v_snap_before.checkpoint_time IS NOT NULL THEN
        IF v_snap_before.checkpoint_time BETWEEN v_window_start AND v_window_end THEN
            v_event := jsonb_build_object(
                'type', 'checkpoint',
                'time', v_snap_before.checkpoint_time,
                'offset_secs', EXTRACT(EPOCH FROM (v_snap_before.checkpoint_time - p_timestamp))::INTEGER
            );
            v_events := v_events || v_event;
            v_notes := array_append(v_notes, format('Checkpoint at %s provides anchor',
                to_char(v_snap_before.checkpoint_time, 'HH24:MI:SS')));
        END IF;
    END IF;

    -- Check for archiver activity
    IF v_snap_before IS NOT NULL AND v_snap_before.last_archived_time IS NOT NULL THEN
        IF v_snap_before.last_archived_time BETWEEN v_window_start AND v_window_end THEN
            v_event := jsonb_build_object(
                'type', 'wal_archived',
                'time', v_snap_before.last_archived_time,
                'offset_secs', EXTRACT(EPOCH FROM (v_snap_before.last_archived_time - p_timestamp))::INTEGER,
                'wal_file', v_snap_before.last_archived_wal
            );
            v_events := v_events || v_event;
        END IF;
    END IF;

    -- Check for archiver failure
    IF v_snap_before IS NOT NULL AND v_snap_before.last_failed_time IS NOT NULL THEN
        IF v_snap_before.last_failed_time BETWEEN v_window_start AND v_window_end THEN
            v_event := jsonb_build_object(
                'type', 'archive_failed',
                'time', v_snap_before.last_failed_time,
                'offset_secs', EXTRACT(EPOCH FROM (v_snap_before.last_failed_time - p_timestamp))::INTEGER,
                'wal_file', v_snap_before.last_failed_wal
            );
            v_events := v_events || v_event;
            v_recs := array_append(v_recs, 'Investigate WAL archiving failure');
        END IF;
    END IF;

    -- ==========================================================================
    -- STEP 5: Analyze activity from samples ring buffer (legacy) or v2
    -- ==========================================================================
    IF v_sample_before IS NOT NULL THEN
        -- Count active sessions
        SELECT COUNT(*), COUNT(*) FILTER (WHERE a.state = 'active')
        INTO v_sessions, v_sessions
        FROM pgfr_record.activity_samples_ring a
        WHERE a.slot_id = v_sample_before.slot_id
          AND a.pid IS NOT NULL;

        -- Find long-running queries (> 60 seconds at sample time)
        SELECT COUNT(*), MAX(EXTRACT(EPOCH FROM (v_sample_before.captured_at - a.query_start)))
        INTO v_long_running, v_longest_secs
        FROM pgfr_record.activity_samples_ring a
        WHERE a.slot_id = v_sample_before.slot_id
          AND a.pid IS NOT NULL
          AND a.state = 'active'
          AND a.query_start IS NOT NULL
          AND a.query_start < v_sample_before.captured_at - interval '60 seconds';

        -- Collect query start events within window
        FOR v_event IN
            SELECT jsonb_build_object(
                'type', 'query_started',
                'time', a.query_start,
                'offset_secs', EXTRACT(EPOCH FROM (a.query_start - p_timestamp))::INTEGER,
                'pid', a.pid,
                'user', a.usename,
                'query_preview', a.query_preview
            )
            FROM pgfr_record.activity_samples_ring a
            WHERE a.slot_id = v_sample_before.slot_id
              AND a.pid IS NOT NULL
              AND a.query_start BETWEEN v_window_start AND v_window_end
            ORDER BY a.query_start
            LIMIT 10
        LOOP
            v_events := v_events || v_event;
        END LOOP;

        -- Collect transaction start events within window
        FOR v_event IN
            SELECT jsonb_build_object(
                'type', 'transaction_started',
                'time', a.xact_start,
                'offset_secs', EXTRACT(EPOCH FROM (a.xact_start - p_timestamp))::INTEGER,
                'pid', a.pid,
                'user', a.usename
            )
            FROM pgfr_record.activity_samples_ring a
            WHERE a.slot_id = v_sample_before.slot_id
              AND a.pid IS NOT NULL
              AND a.xact_start BETWEEN v_window_start AND v_window_end
              AND a.xact_start != a.query_start  -- Avoid duplicates
            ORDER BY a.xact_start
            LIMIT 10
        LOOP
            v_events := v_events || v_event;
        END LOOP;
    ELSIF v_v2_sample_ts_before IS NOT NULL THEN
        -- v2 fallback: activity_samples
        SELECT count(*) FILTER (WHERE a.state = 'active'),
               count(*) FILTER (WHERE a.state = 'active')
        INTO v_sessions, v_sessions
        FROM pgfr_record.activity_samples a
        WHERE a.sample_ts = v_v2_sample_ts_before
          AND a.pid IS NOT NULL;

        SELECT count(*),
               max(extract(epoch from (v_v2_captured_before - a.query_start)))
        INTO v_long_running, v_longest_secs
        FROM pgfr_record.activity_samples a
        WHERE a.sample_ts = v_v2_sample_ts_before
          AND a.pid IS NOT NULL
          AND a.state = 'active'
          AND a.query_start IS NOT NULL
          AND a.query_start < v_v2_captured_before - interval '60 seconds';

        FOR v_event IN
            SELECT jsonb_build_object(
                'type', 'query_started',
                'time', a.query_start,
                'offset_secs', extract(epoch from (a.query_start - p_timestamp))::integer,
                'pid', a.pid,
                'user', a.usename,
                'query_preview', a.query_preview
            )
            FROM pgfr_record.activity_samples a
            WHERE a.sample_ts = v_v2_sample_ts_before
              AND a.pid IS NOT NULL
              AND a.query_start BETWEEN v_window_start AND v_window_end
            ORDER BY a.query_start
            LIMIT 10
        LOOP
            v_events := v_events || v_event;
        END LOOP;

        FOR v_event IN
            SELECT jsonb_build_object(
                'type', 'transaction_started',
                'time', a.xact_start,
                'offset_secs', extract(epoch from (a.xact_start - p_timestamp))::integer,
                'pid', a.pid,
                'user', a.usename
            )
            FROM pgfr_record.activity_samples a
            WHERE a.sample_ts = v_v2_sample_ts_before
              AND a.pid IS NOT NULL
              AND a.xact_start BETWEEN v_window_start AND v_window_end
              AND a.xact_start IS DISTINCT FROM a.query_start
            ORDER BY a.xact_start
            LIMIT 10
        LOOP
            v_events := v_events || v_event;
        END LOOP;
    END IF;

    -- ==========================================================================
    -- STEP 6: Analyze lock contention
    -- ==========================================================================
    IF v_sample_before IS NOT NULL THEN
        SELECT COUNT(*) > 0, COUNT(*)
        INTO v_lock_detected, v_blocked
        FROM pgfr_record.lock_samples_ring l
        WHERE l.slot_id = v_sample_before.slot_id
          AND l.blocked_pid IS NOT NULL;

        IF v_blocked > 0 THEN
            v_recs := array_append(v_recs, format('Investigate %s blocked sessions', v_blocked));
        END IF;
    ELSIF v_v2_sample_ts_before IS NOT NULL THEN
        SELECT count(*) > 0, count(*)
        INTO v_lock_detected, v_blocked
        FROM pgfr_record.lock_samples ls
        WHERE ls.sample_ts = v_v2_sample_ts_before
          AND ls.blocked_pid IS NOT NULL;

        IF v_blocked > 0 THEN
            v_recs := array_append(v_recs, format('Investigate %s blocked sessions', v_blocked));
        END IF;
    END IF;

    -- ==========================================================================
    -- STEP 7: Analyze wait events
    -- ==========================================================================
    IF v_sample_before IS NOT NULL THEN
        SELECT jsonb_agg(w ORDER BY w->>'count' DESC)
        INTO v_waits
        FROM (
            SELECT jsonb_build_object(
                'wait_event_type', ws.wait_event_type,
                'wait_event', ws.wait_event,
                'count', ws.count
            ) AS w
            FROM pgfr_record.wait_samples_ring ws
            WHERE ws.slot_id = v_sample_before.slot_id
              AND ws.wait_event IS NOT NULL
            ORDER BY ws.count DESC NULLS LAST
            LIMIT 5
        ) sub;
    ELSIF v_v2_sample_ts_before IS NOT NULL THEN
        -- v2: decode integer array via wait_event_map
        SELECT jsonb_agg(w ORDER BY w->>'count' DESC)
        INTO v_waits
        FROM (
            SELECT jsonb_build_object(
                'wait_event_type', wem.type,
                'wait_event', wem.event,
                'count', abs(ws.data[i + 1])
            ) AS w
            FROM pgfr_record.wait_samples ws
            cross join generate_subscripts(ws.data, 1) AS i
            join pgfr_record.wait_event_map wem ON wem.id = abs(ws.data[i])::smallint
            WHERE ws.data[i] < 0
              AND ws.sample_ts = v_v2_sample_ts_before
            ORDER BY abs(ws.data[i + 1]) DESC NULLS LAST
            LIMIT 5
        ) sub;
    END IF;

    -- ==========================================================================
    -- STEP 8: Calculate confidence score
    -- ==========================================================================
    -- Base score on sample gap
    IF v_sample_gap_secs IS NOT NULL THEN
        IF v_sample_gap_secs < 60 THEN
            v_confidence_score := 0.9;
        ELSIF v_sample_gap_secs < 300 THEN
            v_confidence_score := 0.7 + (0.2 * (300 - v_sample_gap_secs) / 240);
        ELSIF v_sample_gap_secs < 600 THEN
            v_confidence_score := 0.5 + (0.2 * (600 - v_sample_gap_secs) / 300);
        ELSE
            v_confidence_score := 0.3;
        END IF;

        v_notes := array_append(v_notes, format('Sample gap is %s seconds', v_sample_gap_secs::INTEGER));
    ELSE
        v_confidence_score := 0.2;
        v_notes := array_append(v_notes, 'No sample data in ring buffer');
    END IF;

    -- Bonus for exact-timestamp events
    IF jsonb_array_length(v_events) > 0 THEN
        v_confidence_score := LEAST(1.0, v_confidence_score + 0.1);
        v_notes := array_append(v_notes, format('%s exact-timestamp events found', jsonb_array_length(v_events)));
    END IF;

    -- Bonus for target close to sample
    IF v_sample_before IS NOT NULL OR v_v2_sample_ts_before IS NOT NULL THEN
        DECLARE
            v_closest_gap NUMERIC;
        BEGIN
            v_closest_gap := LEAST(
                ABS(EXTRACT(EPOCH FROM (p_timestamp - coalesce(v_sample_before.captured_at, v_v2_captured_before)))),
                COALESCE(ABS(EXTRACT(EPOCH FROM (p_timestamp - coalesce(v_sample_after.captured_at, v_v2_captured_after)))), 999999)
            );
            IF v_closest_gap < 30 THEN
                v_confidence_score := LEAST(1.0, v_confidence_score + 0.05);
            END IF;
        END;
    END IF;

    -- Determine confidence level
    IF v_confidence_score >= 0.8 THEN
        v_confidence_level := 'high';
    ELSIF v_confidence_score >= 0.6 THEN
        v_confidence_level := 'medium';
    ELSIF v_confidence_score >= 0.4 THEN
        v_confidence_level := 'low';
    ELSE
        v_confidence_level := 'very_low';
    END IF;

    -- ==========================================================================
    -- STEP 9: Generate recommendations
    -- ==========================================================================
    IF v_long_running > 0 THEN
        v_recs := array_append(v_recs, format('Review %s long-running queries (longest: %s sec)',
            v_long_running, round(v_longest_secs)));
    END IF;

    IF v_est_hit_ratio IS NOT NULL AND v_est_hit_ratio < 95 THEN
        v_recs := array_append(v_recs, format('Buffer hit ratio low (%s%%) - consider increasing shared_buffers',
            v_est_hit_ratio));
    END IF;

    -- Check for checkpoint impact
    FOR v_event IN SELECT * FROM jsonb_array_elements(v_events)
    LOOP
        IF v_event->>'type' = 'checkpoint' THEN
            v_recs := array_append(v_recs, 'Review checkpoint impact on performance');
        END IF;
    END LOOP;

    IF array_length(v_recs, 1) IS NULL THEN
        v_recs := array_append(v_recs, 'No immediate concerns detected');
    END IF;

    -- ==========================================================================
    -- RETURN RESULTS
    -- ==========================================================================
    RETURN QUERY SELECT
        p_timestamp,
        coalesce(v_sample_before.captured_at, v_v2_captured_before),
        coalesce(v_sample_after.captured_at,  v_v2_captured_after),
        v_snap_before.captured_at,
        v_snap_after.captured_at,
        v_est_active,
        v_est_total,
        v_est_xact_rate,
        v_est_hit_ratio,
        COALESCE(v_events, '[]'::jsonb),
        v_sessions,
        COALESCE(v_long_running, 0),
        COALESCE(v_longest_secs, 0),
        v_lock_detected,
        v_blocked,
        COALESCE(v_waits, '[]'::jsonb),
        v_confidence_level,
        round(v_confidence_score, 2),
        v_notes,
        v_recs;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.what_happened_at IS
'Time-travel debugging: Forensic analysis of system state at any timestamp. Interpolates between samples to estimate connections, transaction rates, and buffer hit ratio. Surfaces exact-timestamp events (checkpoints, query starts, transaction starts) and analyzes sessions, locks, and wait events. Returns confidence score (0-1) based on data proximity. Use for incident investigation: SELECT * FROM pgfr_analyze.what_happened_at(''2024-01-15 10:23:47'');';


-- Timeline reconstruction for incident analysis
-- Merges events from multiple sources into a unified, chronological timeline
-- Input: Start and end timestamps for the incident window
-- Output: Ordered timeline of events with type, description, and details
CREATE OR REPLACE FUNCTION pgfr_analyze.incident_timeline(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TABLE(
    event_time TIMESTAMPTZ,
    event_type TEXT,
    description TEXT,
    details JSONB
)
LANGUAGE plpgsql AS $$
BEGIN
    RETURN QUERY
    WITH all_events AS (
        -- Checkpoint events from snapshots
        SELECT DISTINCT
            s.checkpoint_time AS event_time,
            'checkpoint'::TEXT AS event_type,
            format('Checkpoint completed (LSN: %s)', s.checkpoint_lsn::TEXT) AS description,
            jsonb_build_object(
                'checkpoint_lsn', s.checkpoint_lsn::TEXT,
                'buffers_written', s.ckpt_buffers,
                'write_time_ms', round(s.ckpt_write_time::NUMERIC, 1),
                'sync_time_ms', round(s.ckpt_sync_time::NUMERIC, 1)
            ) AS details
        FROM pgfr_record.snapshots s
        WHERE s.checkpoint_time BETWEEN p_start_time AND p_end_time
          AND s.checkpoint_time IS NOT NULL

        UNION ALL

        -- WAL archive events
        SELECT DISTINCT
            s.last_archived_time AS event_time,
            'wal_archived'::TEXT AS event_type,
            format('WAL file archived: %s', s.last_archived_wal) AS description,
            jsonb_build_object(
                'wal_file', s.last_archived_wal,
                'archived_count', s.archived_count
            ) AS details
        FROM pgfr_record.snapshots s
        WHERE s.last_archived_time BETWEEN p_start_time AND p_end_time
          AND s.last_archived_time IS NOT NULL

        UNION ALL

        -- WAL archive failures
        SELECT DISTINCT
            s.last_failed_time AS event_time,
            'archive_failed'::TEXT AS event_type,
            format('WAL archive failed: %s', s.last_failed_wal) AS description,
            jsonb_build_object(
                'wal_file', s.last_failed_wal,
                'failed_count', s.failed_count
            ) AS details
        FROM pgfr_record.snapshots s
        WHERE s.last_failed_time BETWEEN p_start_time AND p_end_time
          AND s.last_failed_time IS NOT NULL

        UNION ALL

        -- Query start events from legacy ring buffer
        SELECT
            a.query_start AS event_time,
            'query_started'::TEXT AS event_type,
            format('Query started by %s (pid %s)', COALESCE(a.usename, 'unknown'), a.pid) AS description,
            jsonb_build_object(
                'pid', a.pid,
                'user', a.usename,
                'application', a.application_name,
                'client_addr', a.client_addr::TEXT,
                'query_preview', a.query_preview
            ) AS details
        FROM pgfr_record.activity_samples_ring a
        JOIN pgfr_record.samples_ring sr ON sr.slot_id = a.slot_id
        WHERE a.query_start BETWEEN p_start_time AND p_end_time
          AND a.query_start IS NOT NULL
          AND a.pid IS NOT NULL
          AND sr.captured_at > '1970-01-01'::timestamptz

        UNION ALL

        -- Query start events from v2 ring
        SELECT
            a.query_start AS event_time,
            'query_started'::TEXT AS event_type,
            format('Query started by %s (pid %s)', COALESCE(a.usename, 'unknown'), a.pid) AS description,
            jsonb_build_object(
                'pid', a.pid,
                'user', a.usename,
                'application', a.application_name,
                'client_addr', a.client_addr::TEXT,
                'query_preview', a.query_preview
            ) AS details
        FROM pgfr_record.activity_samples a
        WHERE a.query_start BETWEEN p_start_time AND p_end_time
          AND a.query_start IS NOT NULL
          AND a.pid IS NOT NULL

        UNION ALL

        -- Transaction start events from legacy ring buffer
        SELECT
            a.xact_start AS event_time,
            'transaction_started'::TEXT AS event_type,
            format('Transaction started by %s (pid %s)', COALESCE(a.usename, 'unknown'), a.pid) AS description,
            jsonb_build_object(
                'pid', a.pid,
                'user', a.usename,
                'application', a.application_name
            ) AS details
        FROM pgfr_record.activity_samples_ring a
        JOIN pgfr_record.samples_ring sr ON sr.slot_id = a.slot_id
        WHERE a.xact_start BETWEEN p_start_time AND p_end_time
          AND a.xact_start IS NOT NULL
          AND a.pid IS NOT NULL
          AND sr.captured_at > '1970-01-01'::timestamptz
          AND (a.xact_start != a.query_start OR a.query_start IS NULL)

        UNION ALL

        -- Transaction start events from v2 ring
        SELECT
            a.xact_start AS event_time,
            'transaction_started'::TEXT AS event_type,
            format('Transaction started by %s (pid %s)', COALESCE(a.usename, 'unknown'), a.pid) AS description,
            jsonb_build_object(
                'pid', a.pid,
                'user', a.usename,
                'application', a.application_name
            ) AS details
        FROM pgfr_record.activity_samples a
        WHERE a.xact_start BETWEEN p_start_time AND p_end_time
          AND a.xact_start IS NOT NULL
          AND a.pid IS NOT NULL
          AND a.xact_start IS DISTINCT FROM a.query_start

        UNION ALL

        -- Backend start events (new connections) from legacy ring
        SELECT
            a.backend_start AS event_time,
            'connection_opened'::TEXT AS event_type,
            format('Connection opened by %s from %s', COALESCE(a.usename, 'unknown'),
                   COALESCE(a.client_addr::TEXT, 'local')) AS description,
            jsonb_build_object(
                'pid', a.pid,
                'user', a.usename,
                'application', a.application_name,
                'client_addr', a.client_addr::TEXT,
                'backend_type', a.backend_type
            ) AS details
        FROM pgfr_record.activity_samples_ring a
        JOIN pgfr_record.samples_ring sr ON sr.slot_id = a.slot_id
        WHERE a.backend_start BETWEEN p_start_time AND p_end_time
          AND a.backend_start IS NOT NULL
          AND a.pid IS NOT NULL
          AND sr.captured_at > '1970-01-01'::timestamptz

        UNION ALL

        -- Backend start events from v2 ring
        SELECT
            a.backend_start AS event_time,
            'connection_opened'::TEXT AS event_type,
            format('Connection opened by %s from %s', COALESCE(a.usename, 'unknown'),
                   COALESCE(a.client_addr::TEXT, 'local')) AS description,
            jsonb_build_object(
                'pid', a.pid,
                'user', a.usename,
                'application', a.application_name,
                'client_addr', a.client_addr::TEXT,
                'backend_type', a.backend_type
            ) AS details
        FROM pgfr_record.activity_samples a
        WHERE a.backend_start BETWEEN p_start_time AND p_end_time
          AND a.backend_start IS NOT NULL
          AND a.pid IS NOT NULL

        UNION ALL

        -- Lock contention events from legacy ring
        SELECT
            sr.captured_at AS event_time,
            'lock_contention'::TEXT AS event_type,
            format('Session %s blocked by %s on %s lock',
                   l.blocked_pid, l.blocking_pid, l.lock_type) AS description,
            jsonb_build_object(
                'blocked_pid', l.blocked_pid,
                'blocked_user', l.blocked_user,
                'blocked_query', l.blocked_query_preview,
                'blocking_pid', l.blocking_pid,
                'blocking_user', l.blocking_user,
                'blocking_query', l.blocking_query_preview,
                'lock_type', l.lock_type,
                'duration', l.blocked_duration::TEXT
            ) AS details
        FROM pgfr_record.lock_samples_ring l
        JOIN pgfr_record.samples_ring sr ON sr.slot_id = l.slot_id
        WHERE sr.captured_at BETWEEN p_start_time AND p_end_time
          AND l.blocked_pid IS NOT NULL
          AND sr.captured_at > '1970-01-01'::timestamptz

        UNION ALL

        -- Lock contention events from v2 ring
        SELECT
            pgfr_record.epoch() + ls.sample_ts * interval '1 second' AS event_time,
            'lock_contention'::TEXT AS event_type,
            format('Session %s blocked by %s on %s lock',
                   ls.blocked_pid, ls.blocking_pid,
                   coalesce(ltm.lock_type, ls.lock_type::text)) AS description,
            jsonb_build_object(
                'blocked_pid', ls.blocked_pid,
                'blocking_pid', ls.blocking_pid,
                'lock_type', coalesce(ltm.lock_type, ls.lock_type::text),
                'duration_s', ls.blocked_duration_s
            ) AS details
        FROM pgfr_record.lock_samples ls
        LEFT JOIN pgfr_record.lock_type_map ltm ON ltm.id = ls.lock_type
        WHERE pgfr_record.epoch() + ls.sample_ts * interval '1 second'
              BETWEEN p_start_time AND p_end_time
          AND ls.blocked_pid IS NOT NULL

        UNION ALL

        -- Wait event spikes from aggregates
        SELECT
            wa.start_time AS event_time,
            'wait_spike'::TEXT AS event_type,
            format('Wait spike: %s/%s (max %s concurrent)',
                   wa.wait_event_type, wa.wait_event, wa.max_waiters) AS description,
            jsonb_build_object(
                'wait_event_type', wa.wait_event_type,
                'wait_event', wa.wait_event,
                'max_concurrent', wa.max_waiters,
                'avg_concurrent', round(wa.avg_waiters, 1),
                'sample_count', wa.sample_count
            ) AS details
        FROM pgfr_record.wait_event_aggregates wa
        WHERE wa.start_time BETWEEN p_start_time AND p_end_time
          AND wa.max_waiters >= 3  -- Only show significant waits

        UNION ALL

        -- Snapshot captures (system state markers)
        SELECT
            s.captured_at AS event_time,
            'snapshot'::TEXT AS event_type,
            format('System snapshot: %s active connections, %s TPS',
                   s.connections_active,
                   CASE WHEN lag(s.xact_commit) OVER (ORDER BY s.captured_at) IS NOT NULL
                        THEN round((s.xact_commit - lag(s.xact_commit) OVER (ORDER BY s.captured_at))::NUMERIC /
                                   NULLIF(EXTRACT(EPOCH FROM (s.captured_at - lag(s.captured_at) OVER (ORDER BY s.captured_at))), 0), 1)
                        ELSE NULL
                   END) AS description,
            jsonb_build_object(
                'connections_active', s.connections_active,
                'connections_total', s.connections_total,
                'xact_commit', s.xact_commit,
                'blks_hit', s.blks_hit,
                'blks_read', s.blks_read,
                'temp_bytes', s.temp_bytes
            ) AS details
        FROM pgfr_record.snapshots s
        WHERE s.captured_at BETWEEN p_start_time AND p_end_time
    )
    SELECT ae.event_time, ae.event_type, ae.description, ae.details
    FROM all_events ae
    WHERE ae.event_time IS NOT NULL
    ORDER BY ae.event_time;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.incident_timeline IS
'Reconstructs a unified timeline for incident analysis by merging events from multiple sources: checkpoints, WAL archiving, query/transaction starts, connection opens, lock contention, wait spikes, and snapshots. Returns chronologically ordered events with type, description, and JSON details. Use for incident review: SELECT * FROM pgfr_analyze.incident_timeline(now() - interval ''2 hours'', now() - interval ''1 hour'');';


-- Blast Radius Analysis
-- Comprehensive impact assessment of database incidents
-- Answers: "What was the collateral damage from this incident?"
-- Input: Start and end timestamps for the incident window
-- Output: Structured impact assessment including locks, queries, connections, applications
CREATE OR REPLACE FUNCTION pgfr_analyze.blast_radius(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TABLE(
    -- Time window
    incident_start TIMESTAMPTZ,
    incident_end TIMESTAMPTZ,
    duration_seconds NUMERIC,

    -- Lock impact
    blocked_sessions_total INTEGER,
    blocked_sessions_max_concurrent INTEGER,
    max_block_duration INTERVAL,
    avg_block_duration INTERVAL,
    lock_types JSONB,

    -- Query degradation
    degraded_queries_count INTEGER,
    degraded_queries JSONB,

    -- Connection impact
    connections_before INTEGER,
    connections_during_avg INTEGER,
    connections_during_max INTEGER,
    connection_increase_pct NUMERIC,

    -- Application impact
    affected_applications JSONB,

    -- Wait event impact
    top_wait_events JSONB,

    -- Transaction throughput
    tps_before NUMERIC,
    tps_during NUMERIC,
    tps_change_pct NUMERIC,

    -- Summary
    severity TEXT,
    impact_summary TEXT[],
    recommendations TEXT[]
)
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_duration INTERVAL;
    v_baseline_start TIMESTAMPTZ;
    v_baseline_end TIMESTAMPTZ;

    -- Lock impact variables
    v_blocked_total INTEGER := 0;
    v_blocked_max_concurrent INTEGER := 0;
    v_max_block_duration INTERVAL;
    v_avg_block_duration INTERVAL;
    v_lock_types JSONB := '[]'::jsonb;

    -- Query degradation variables
    v_degraded_count INTEGER := 0;
    v_degraded_queries JSONB := '[]'::jsonb;

    -- Connection variables
    v_conn_before INTEGER := 0;
    v_conn_during_avg INTEGER := 0;
    v_conn_during_max INTEGER := 0;
    v_conn_increase_pct NUMERIC := 0;

    -- Application impact
    v_affected_apps JSONB := '[]'::jsonb;

    -- Wait events
    v_top_waits JSONB := '[]'::jsonb;

    -- Throughput variables
    v_tps_before NUMERIC := 0;
    v_tps_during NUMERIC := 0;
    v_tps_change_pct NUMERIC := 0;

    -- Severity calculation
    v_severity TEXT := 'low';
    v_impact_summary TEXT[] := ARRAY[]::TEXT[];
    v_recommendations TEXT[] := ARRAY[]::TEXT[];

    -- Severity scores
    v_lock_severity INTEGER := 0;
    v_duration_severity INTEGER := 0;
    v_conn_severity INTEGER := 0;
    v_tps_severity INTEGER := 0;
    v_query_severity INTEGER := 0;
BEGIN
    -- Calculate duration and baseline period
    v_duration := p_end_time - p_start_time;
    v_baseline_start := p_start_time - v_duration;
    v_baseline_end := p_start_time;

    -- =========================================================================
    -- Lock Impact Analysis
    -- =========================================================================

    -- Total blocked sessions and durations from ring buffer
    SELECT
        COUNT(DISTINCT blocked_pid),
        MAX(blocked_duration),
        AVG(blocked_duration)
    INTO v_blocked_total, v_max_block_duration, v_avg_block_duration
    FROM (
        SELECT l.blocked_pid, l.blocked_duration
        FROM pgfr_record.lock_samples_ring l
        JOIN pgfr_record.samples_ring sr ON sr.slot_id = l.slot_id
        WHERE sr.captured_at BETWEEN p_start_time AND p_end_time
          AND l.blocked_pid IS NOT NULL
        UNION ALL
        SELECT ls.blocked_pid,
               ls.blocked_duration_s * interval '1 second'
        FROM pgfr_record.lock_samples ls
        WHERE pgfr_record.epoch() + ls.sample_ts * interval '1 second'
              BETWEEN p_start_time AND p_end_time
          AND ls.blocked_pid IS NOT NULL
    ) combined;

    -- Also check archive for longer incidents
    IF v_blocked_total = 0 OR v_blocked_total IS NULL THEN
        SELECT
            COUNT(DISTINCT blocked_pid),
            MAX(blocked_duration),
            AVG(blocked_duration)
        INTO v_blocked_total, v_max_block_duration, v_avg_block_duration
        FROM pgfr_record.lock_samples_archive
        WHERE captured_at BETWEEN p_start_time AND p_end_time
          AND blocked_pid IS NOT NULL;
    END IF;

    v_blocked_total := COALESCE(v_blocked_total, 0);

    -- Max concurrent blocked sessions (per sample)
    SELECT COALESCE(MAX(blocked_count), 0)
    INTO v_blocked_max_concurrent
    FROM (
        SELECT sample_key, COUNT(DISTINCT blocked_pid) AS blocked_count
        FROM (
            SELECT sr.slot_id::text AS sample_key, l.blocked_pid
            FROM pgfr_record.lock_samples_ring l
            JOIN pgfr_record.samples_ring sr ON sr.slot_id = l.slot_id
            WHERE sr.captured_at BETWEEN p_start_time AND p_end_time
              AND l.blocked_pid IS NOT NULL
            UNION ALL
            SELECT ls.sample_ts::text AS sample_key, ls.blocked_pid
            FROM pgfr_record.lock_samples ls
            WHERE pgfr_record.epoch() + ls.sample_ts * interval '1 second'
                  BETWEEN p_start_time AND p_end_time
              AND ls.blocked_pid IS NOT NULL
        ) combined
        GROUP BY sample_key
    ) per_sample;

    -- Lock types breakdown
    SELECT COALESCE(jsonb_agg(jsonb_build_object('type', lock_type, 'count', cnt) ORDER BY cnt DESC), '[]'::jsonb)
    INTO v_lock_types
    FROM (
        SELECT lock_type, COUNT(*) AS cnt
        FROM (
            SELECT l.lock_type
            FROM pgfr_record.lock_samples_ring l
            JOIN pgfr_record.samples_ring sr ON sr.slot_id = l.slot_id
            WHERE sr.captured_at BETWEEN p_start_time AND p_end_time
              AND l.blocked_pid IS NOT NULL
              AND l.lock_type IS NOT NULL
            UNION ALL
            SELECT coalesce(ltm.lock_type, ls.lock_type::text) AS lock_type
            FROM pgfr_record.lock_samples ls
            LEFT JOIN pgfr_record.lock_type_map ltm ON ltm.id = ls.lock_type
            WHERE pgfr_record.epoch() + ls.sample_ts * interval '1 second'
                  BETWEEN p_start_time AND p_end_time
              AND ls.blocked_pid IS NOT NULL
        ) combined
        GROUP BY lock_type
        ORDER BY cnt DESC
        LIMIT 10
    ) lt;

    -- =========================================================================
    -- Query Degradation Analysis
    -- =========================================================================

    WITH baseline AS (
        SELECT
            ss.queryid,
            left(ss.query_preview, 80) AS query_preview,
            AVG(ss.mean_exec_time) AS baseline_ms
        FROM pgfr_record.statement_snapshots ss
        JOIN pgfr_record.snapshots s ON s.id = ss.snapshot_id
        WHERE s.captured_at BETWEEN v_baseline_start AND v_baseline_end
          AND ss.mean_exec_time IS NOT NULL
          AND ss.mean_exec_time > 0
        GROUP BY ss.queryid, left(ss.query_preview, 80)
        HAVING COUNT(*) >= 1
    ),
    during AS (
        SELECT
            ss.queryid,
            AVG(ss.mean_exec_time) AS during_ms
        FROM pgfr_record.statement_snapshots ss
        JOIN pgfr_record.snapshots s ON s.id = ss.snapshot_id
        WHERE s.captured_at BETWEEN p_start_time AND p_end_time
          AND ss.mean_exec_time IS NOT NULL
          AND ss.mean_exec_time > 0
        GROUP BY ss.queryid
        HAVING COUNT(*) >= 1
    ),
    degraded AS (
        SELECT
            b.queryid,
            b.query_preview,
            round(b.baseline_ms::numeric, 2) AS baseline_ms,
            round(d.during_ms::numeric, 2) AS during_ms,
            round((100.0 * (d.during_ms - b.baseline_ms) / NULLIF(b.baseline_ms, 0))::numeric, 1) AS slowdown_pct
        FROM baseline b
        JOIN during d ON d.queryid = b.queryid
        WHERE d.during_ms > b.baseline_ms * 1.5  -- 50%+ slower
        ORDER BY (d.during_ms - b.baseline_ms) DESC
        LIMIT 10
    )
    SELECT
        COUNT(*)::integer,
        COALESCE(jsonb_agg(jsonb_build_object(
            'queryid', queryid,
            'query_preview', query_preview,
            'baseline_ms', baseline_ms,
            'during_ms', during_ms,
            'slowdown_pct', slowdown_pct
        )), '[]'::jsonb)
    INTO v_degraded_count, v_degraded_queries
    FROM degraded;

    v_degraded_count := COALESCE(v_degraded_count, 0);

    -- =========================================================================
    -- Connection Impact Analysis
    -- =========================================================================

    -- Baseline connections (average before incident)
    SELECT COALESCE(AVG(connections_total), 0)::integer
    INTO v_conn_before
    FROM pgfr_record.snapshots
    WHERE captured_at BETWEEN v_baseline_start AND v_baseline_end;

    -- During incident connections
    SELECT
        COALESCE(AVG(connections_total), 0)::integer,
        COALESCE(MAX(connections_total), 0)::integer
    INTO v_conn_during_avg, v_conn_during_max
    FROM pgfr_record.snapshots
    WHERE captured_at BETWEEN p_start_time AND p_end_time;

    -- Connection increase percentage
    IF v_conn_before > 0 THEN
        v_conn_increase_pct := round(100.0 * (v_conn_during_avg - v_conn_before) / v_conn_before, 1);
    END IF;

    -- =========================================================================
    -- Application Impact Analysis
    -- =========================================================================

    SELECT COALESCE(jsonb_agg(jsonb_build_object(
        'app_name', app,
        'blocked_count', blocked_count,
        'max_wait', max_wait
    ) ORDER BY blocked_count DESC), '[]'::jsonb)
    INTO v_affected_apps
    FROM (
        SELECT
            app,
            COUNT(DISTINCT blocked_pid) AS blocked_count,
            MAX(blocked_duration) AS max_wait
        FROM (
            SELECT COALESCE(l.blocked_app, 'unknown') AS app,
                   l.blocked_pid,
                   l.blocked_duration
            FROM pgfr_record.lock_samples_ring l
            JOIN pgfr_record.samples_ring sr ON sr.slot_id = l.slot_id
            WHERE sr.captured_at BETWEEN p_start_time AND p_end_time
              AND l.blocked_pid IS NOT NULL
            UNION ALL
            SELECT 'unknown' AS app,
                   ls.blocked_pid,
                   ls.blocked_duration_s * interval '1 second'
            FROM pgfr_record.lock_samples ls
            WHERE pgfr_record.epoch() + ls.sample_ts * interval '1 second'
                  BETWEEN p_start_time AND p_end_time
              AND ls.blocked_pid IS NOT NULL
        ) combined
        GROUP BY app
        ORDER BY blocked_count DESC
        LIMIT 10
    ) apps;

    -- =========================================================================
    -- Wait Event Analysis
    -- =========================================================================

    WITH baseline_waits AS (
        SELECT wait_event_type, wait_event, SUM(total_count) AS total_count
        FROM (
            -- legacy ring
            SELECT w.wait_event_type, w.wait_event, w.count::bigint AS total_count
            FROM pgfr_record.wait_samples_ring w
            JOIN pgfr_record.samples_ring sr ON sr.slot_id = w.slot_id
            WHERE sr.captured_at BETWEEN v_baseline_start AND v_baseline_end
              AND w.wait_event IS NOT NULL
            UNION ALL
            -- v2 ring: decode integer array
            SELECT wem.type AS wait_event_type, wem.event AS wait_event,
                   abs(ws.data[i + 1])::bigint AS total_count
            FROM pgfr_record.wait_samples ws
            cross join generate_subscripts(ws.data, 1) AS i
            join pgfr_record.wait_event_map wem ON wem.id = abs(ws.data[i])::smallint
            WHERE ws.data[i] < 0
              AND pgfr_record.epoch() + ws.sample_ts * interval '1 second'
                  BETWEEN v_baseline_start AND v_baseline_end
        ) combined
        GROUP BY wait_event_type, wait_event
    ),
    during_waits AS (
        SELECT wait_event_type, wait_event, SUM(total_count) AS total_count
        FROM (
            -- legacy ring
            SELECT w.wait_event_type, w.wait_event, w.count::bigint AS total_count
            FROM pgfr_record.wait_samples_ring w
            JOIN pgfr_record.samples_ring sr ON sr.slot_id = w.slot_id
            WHERE sr.captured_at BETWEEN p_start_time AND p_end_time
              AND w.wait_event IS NOT NULL
            UNION ALL
            -- v2 ring: decode integer array
            SELECT wem.type AS wait_event_type, wem.event AS wait_event,
                   abs(ws.data[i + 1])::bigint AS total_count
            FROM pgfr_record.wait_samples ws
            cross join generate_subscripts(ws.data, 1) AS i
            join pgfr_record.wait_event_map wem ON wem.id = abs(ws.data[i])::smallint
            WHERE ws.data[i] < 0
              AND pgfr_record.epoch() + ws.sample_ts * interval '1 second'
                  BETWEEN p_start_time AND p_end_time
        ) combined
        GROUP BY wait_event_type, wait_event
    )
    SELECT COALESCE(jsonb_agg(jsonb_build_object(
        'wait_type', d.wait_event_type,
        'wait_event', d.wait_event,
        'total_count', d.total_count,
        'pct_increase', CASE
            WHEN COALESCE(b.total_count, 0) > 0
            THEN round(100.0 * (d.total_count - COALESCE(b.total_count, 0)) / b.total_count, 1)
            ELSE NULL
        END
    ) ORDER BY d.total_count DESC), '[]'::jsonb)
    INTO v_top_waits
    FROM during_waits d
    LEFT JOIN baseline_waits b ON b.wait_event_type = d.wait_event_type
                               AND b.wait_event = d.wait_event
    WHERE d.total_count > 0
    LIMIT 10;

    -- =========================================================================
    -- Transaction Throughput Analysis
    -- =========================================================================

    -- Calculate TPS before incident
    WITH baseline_tps AS (
        SELECT
            xact_commit,
            captured_at,
            LAG(xact_commit) OVER (ORDER BY captured_at) AS prev_commit,
            LAG(captured_at) OVER (ORDER BY captured_at) AS prev_time
        FROM pgfr_record.snapshots
        WHERE captured_at BETWEEN v_baseline_start AND v_baseline_end
    )
    SELECT COALESCE(AVG(
        CASE
            WHEN prev_commit IS NOT NULL AND prev_time IS NOT NULL
                 AND EXTRACT(EPOCH FROM (captured_at - prev_time)) > 0
            THEN (xact_commit - prev_commit)::numeric / EXTRACT(EPOCH FROM (captured_at - prev_time))
            ELSE NULL
        END
    ), 0)
    INTO v_tps_before
    FROM baseline_tps;

    -- Calculate TPS during incident
    WITH during_tps AS (
        SELECT
            xact_commit,
            captured_at,
            LAG(xact_commit) OVER (ORDER BY captured_at) AS prev_commit,
            LAG(captured_at) OVER (ORDER BY captured_at) AS prev_time
        FROM pgfr_record.snapshots
        WHERE captured_at BETWEEN p_start_time AND p_end_time
    )
    SELECT COALESCE(AVG(
        CASE
            WHEN prev_commit IS NOT NULL AND prev_time IS NOT NULL
                 AND EXTRACT(EPOCH FROM (captured_at - prev_time)) > 0
            THEN (xact_commit - prev_commit)::numeric / EXTRACT(EPOCH FROM (captured_at - prev_time))
            ELSE NULL
        END
    ), 0)
    INTO v_tps_during
    FROM during_tps;

    -- TPS change percentage
    IF v_tps_before > 0 THEN
        v_tps_change_pct := round(100.0 * (v_tps_during - v_tps_before) / v_tps_before, 1);
    END IF;

    -- =========================================================================
    -- Severity Classification
    -- =========================================================================

    -- Blocked sessions severity: 1-5=low, 6-20=medium, 21-50=high, >50=critical
    v_lock_severity := CASE
        WHEN v_blocked_total > 50 THEN 4
        WHEN v_blocked_total > 20 THEN 3
        WHEN v_blocked_total > 5 THEN 2
        WHEN v_blocked_total > 0 THEN 1
        ELSE 0
    END;

    -- Block duration severity: <10s=low, 10-60s=medium, 1-5min=high, >5min=critical
    v_duration_severity := CASE
        WHEN v_max_block_duration > interval '5 minutes' THEN 4
        WHEN v_max_block_duration > interval '1 minute' THEN 3
        WHEN v_max_block_duration > interval '10 seconds' THEN 2
        WHEN v_max_block_duration IS NOT NULL THEN 1
        ELSE 0
    END;

    -- Connection increase severity: <25%=low, 25-50%=medium, 50-100%=high, >100%=critical
    v_conn_severity := CASE
        WHEN v_conn_increase_pct > 100 THEN 4
        WHEN v_conn_increase_pct > 50 THEN 3
        WHEN v_conn_increase_pct > 25 THEN 2
        WHEN v_conn_increase_pct > 0 THEN 1
        ELSE 0
    END;

    -- TPS decrease severity: <10%=low, 10-25%=medium, 25-50%=high, >50%=critical
    v_tps_severity := CASE
        WHEN v_tps_change_pct < -50 THEN 4
        WHEN v_tps_change_pct < -25 THEN 3
        WHEN v_tps_change_pct < -10 THEN 2
        WHEN v_tps_change_pct < 0 THEN 1
        ELSE 0
    END;

    -- Degraded queries severity: 1-3=low, 4-10=medium, 11-25=high, >25=critical
    v_query_severity := CASE
        WHEN v_degraded_count > 25 THEN 4
        WHEN v_degraded_count > 10 THEN 3
        WHEN v_degraded_count > 3 THEN 2
        WHEN v_degraded_count > 0 THEN 1
        ELSE 0
    END;

    -- Overall severity = highest individual severity
    v_severity := CASE GREATEST(v_lock_severity, v_duration_severity, v_conn_severity, v_tps_severity, v_query_severity)
        WHEN 4 THEN 'critical'
        WHEN 3 THEN 'high'
        WHEN 2 THEN 'medium'
        WHEN 1 THEN 'low'
        ELSE 'low'
    END;

    -- =========================================================================
    -- Impact Summary
    -- =========================================================================

    IF v_blocked_total > 0 THEN
        v_impact_summary := array_append(v_impact_summary,
            format('%s sessions blocked (max %s)',
                   v_blocked_total,
                   COALESCE(to_char(v_max_block_duration, 'MI"m" SS"s"'), 'unknown')));
    END IF;

    IF v_tps_change_pct < -10 THEN
        v_impact_summary := array_append(v_impact_summary,
            format('TPS dropped %s%%', abs(v_tps_change_pct)::integer));
    END IF;

    IF v_degraded_count > 0 THEN
        v_impact_summary := array_append(v_impact_summary,
            format('%s queries degraded >50%%', v_degraded_count));
    END IF;

    IF v_conn_increase_pct > 25 THEN
        v_impact_summary := array_append(v_impact_summary,
            format('Connections increased %s%%', v_conn_increase_pct::integer));
    END IF;

    IF array_length(v_impact_summary, 1) IS NULL THEN
        v_impact_summary := array_append(v_impact_summary, 'No significant impact detected');
    END IF;

    -- =========================================================================
    -- Recommendations
    -- =========================================================================

    IF v_max_block_duration > interval '1 minute' THEN
        v_recommendations := array_append(v_recommendations,
            'Review the blocking query that held locks for extended duration');
    END IF;

    IF v_blocked_total > 10 THEN
        v_recommendations := array_append(v_recommendations,
            'Consider setting lock_timeout to prevent long waits');
    END IF;

    IF v_conn_increase_pct > 50 THEN
        v_recommendations := array_append(v_recommendations,
            format('Investigate connection pool sizing (reached %s connections)', v_conn_during_max));
    END IF;

    IF v_degraded_count > 3 THEN
        v_recommendations := array_append(v_recommendations,
            format('%s queries showed >50%% degradation - review execution plans', v_degraded_count));
    END IF;

    IF v_tps_change_pct < -25 THEN
        v_recommendations := array_append(v_recommendations,
            'Throughput dropped significantly - analyze root cause of slowdown');
    END IF;

    IF array_length(v_recommendations, 1) IS NULL THEN
        v_recommendations := array_append(v_recommendations, 'No specific recommendations');
    END IF;

    -- =========================================================================
    -- Return Results
    -- =========================================================================

    RETURN QUERY SELECT
        p_start_time,
        p_end_time,
        round(EXTRACT(EPOCH FROM v_duration), 0),
        v_blocked_total,
        v_blocked_max_concurrent,
        v_max_block_duration,
        v_avg_block_duration,
        v_lock_types,
        v_degraded_count,
        v_degraded_queries,
        v_conn_before,
        v_conn_during_avg,
        v_conn_during_max,
        v_conn_increase_pct,
        v_affected_apps,
        v_top_waits,
        round(v_tps_before, 1),
        round(v_tps_during, 1),
        v_tps_change_pct,
        v_severity,
        v_impact_summary,
        v_recommendations;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.blast_radius IS
'Comprehensive blast radius analysis for incident impact assessment. Analyzes lock impact (blocked sessions, duration, types), query degradation (before vs during), connection spike, affected applications, wait events, and transaction throughput. Returns severity classification (low/medium/high/critical) with impact summary and recommendations. Use for incident postmortems: SELECT * FROM pgfr_analyze.blast_radius(''2024-01-15 10:23:00'', ''2024-01-15 10:35:00'');';


-- Blast Radius Report
-- Human-readable formatted report for incident postmortems
-- Returns ASCII-art styled report suitable for sharing and documentation
CREATE OR REPLACE FUNCTION pgfr_analyze.blast_radius_report(
    p_start_time TIMESTAMPTZ,
    p_end_time TIMESTAMPTZ
)
RETURNS TEXT
LANGUAGE plpgsql STABLE AS $$
DECLARE
    v_data RECORD;
    v_result TEXT := '';
    v_bar TEXT;
    v_max_bar_width INTEGER := 20;
    v_app RECORD;
    v_query RECORD;
    v_wait RECORD;
    v_lock RECORD;
    v_severity_bar TEXT;
BEGIN
    -- Get blast radius data
    SELECT * INTO v_data FROM pgfr_analyze.blast_radius(p_start_time, p_end_time);

    -- Header
    v_result := v_result || E'══════════════════════════════════════════════════════════════════════\n';
    v_result := v_result || E'                    BLAST RADIUS ANALYSIS REPORT\n';
    v_result := v_result || E'══════════════════════════════════════════════════════════════════════\n';

    -- Time window and severity
    v_result := v_result || format('Time Window: %s → %s (%s)',
        to_char(p_start_time, 'YYYY-MM-DD HH24:MI:SS'),
        to_char(p_end_time, 'HH24:MI:SS'),
        CASE
            WHEN v_data.duration_seconds >= 3600 THEN format('%s hours', round(v_data.duration_seconds / 3600, 1))
            WHEN v_data.duration_seconds >= 60 THEN format('%s minutes', round(v_data.duration_seconds / 60, 0))
            ELSE format('%s seconds', v_data.duration_seconds::integer)
        END
    ) || E'\n';

    -- Severity indicator with visual bar
    v_severity_bar := CASE v_data.severity
        WHEN 'critical' THEN '██████████'
        WHEN 'high' THEN '███████░░░'
        WHEN 'medium' THEN '█████░░░░░'
        ELSE '██░░░░░░░░'
    END;
    v_result := v_result || format('Severity: %s %s', v_severity_bar, upper(v_data.severity)) || E'\n\n';

    -- =========================================================================
    -- Lock Impact Section
    -- =========================================================================
    v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';
    v_result := v_result || E'LOCK IMPACT\n';
    v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';

    IF v_data.blocked_sessions_total > 0 THEN
        v_result := v_result || format('  Total blocked sessions:     %s', v_data.blocked_sessions_total) || E'\n';
        v_result := v_result || format('  Max concurrent blocked:     %s', v_data.blocked_sessions_max_concurrent) || E'\n';
        v_result := v_result || format('  Longest block duration:     %s',
            CASE
                WHEN v_data.max_block_duration >= interval '1 hour'
                THEN format('%sh %sm', EXTRACT(HOUR FROM v_data.max_block_duration)::integer,
                                       EXTRACT(MINUTE FROM v_data.max_block_duration)::integer)
                WHEN v_data.max_block_duration >= interval '1 minute'
                THEN format('%sm %ss', EXTRACT(MINUTE FROM v_data.max_block_duration)::integer,
                                       EXTRACT(SECOND FROM v_data.max_block_duration)::integer)
                ELSE format('%ss', round(EXTRACT(EPOCH FROM v_data.max_block_duration), 1))
            END
        ) || E'\n';
        v_result := v_result || format('  Average block duration:     %s',
            CASE
                WHEN v_data.avg_block_duration >= interval '1 minute'
                THEN format('%sm %ss', EXTRACT(MINUTE FROM v_data.avg_block_duration)::integer,
                                       EXTRACT(SECOND FROM v_data.avg_block_duration)::integer)
                ELSE format('%ss', round(EXTRACT(EPOCH FROM v_data.avg_block_duration), 1))
            END
        ) || E'\n\n';

        -- Lock types with bar chart
        IF jsonb_array_length(v_data.lock_types) > 0 THEN
            v_result := v_result || E'  Lock types:\n';
            FOR v_lock IN SELECT * FROM jsonb_to_recordset(v_data.lock_types) AS x(type text, count integer) LOOP
                v_bar := repeat('█', LEAST(v_lock.count, v_max_bar_width));
                v_result := v_result || format('    %-12s %s %s', v_lock.type, rpad(v_bar, v_max_bar_width), v_lock.count) || E'\n';
            END LOOP;
        END IF;
    ELSE
        v_result := v_result || E'  No lock contention detected.\n';
    END IF;
    v_result := v_result || E'\n';

    -- =========================================================================
    -- Affected Applications Section
    -- =========================================================================
    IF jsonb_array_length(v_data.affected_applications) > 0 THEN
        v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';
        v_result := v_result || E'AFFECTED APPLICATIONS\n';
        v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';

        FOR v_app IN SELECT * FROM jsonb_to_recordset(v_data.affected_applications)
                     AS x(app_name text, blocked_count integer, max_wait interval) LOOP
            v_bar := repeat('█', LEAST(v_app.blocked_count, v_max_bar_width));
            v_result := v_result || format('  %-16s %s %s blocked',
                left(v_app.app_name, 16), rpad(v_bar, v_max_bar_width), v_app.blocked_count) || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- =========================================================================
    -- Query Degradation Section
    -- =========================================================================
    v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';
    v_result := v_result || format('QUERY DEGRADATION (%s queries slowed >50%%)', v_data.degraded_queries_count) || E'\n';
    v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';

    IF v_data.degraded_queries_count > 0 THEN
        FOR v_query IN SELECT * FROM jsonb_to_recordset(v_data.degraded_queries)
                       AS x(queryid bigint, query_preview text, baseline_ms numeric, during_ms numeric, slowdown_pct numeric)
                       LIMIT 5 LOOP
            v_result := v_result || format('  %s', left(v_query.query_preview, 50)) || E'\n';
            v_result := v_result || format('    %sms → %sms  (+%s%%)',
                v_query.baseline_ms, v_query.during_ms, v_query.slowdown_pct::integer) || E'\n';
        END LOOP;
        IF v_data.degraded_queries_count > 5 THEN
            v_result := v_result || format('  ... and %s more', v_data.degraded_queries_count - 5) || E'\n';
        END IF;
    ELSE
        v_result := v_result || E'  No significant query degradation detected.\n';
    END IF;
    v_result := v_result || E'\n';

    -- =========================================================================
    -- Resource Impact Section
    -- =========================================================================
    v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';
    v_result := v_result || E'RESOURCE IMPACT\n';
    v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';

    v_result := v_result || format('  Connections:  %s → %s avg (%s max)  %s',
        v_data.connections_before,
        v_data.connections_during_avg,
        v_data.connections_during_max,
        CASE
            WHEN v_data.connection_increase_pct > 0 THEN format('+%s%%', v_data.connection_increase_pct::integer)
            WHEN v_data.connection_increase_pct < 0 THEN format('%s%%', v_data.connection_increase_pct::integer)
            ELSE 'unchanged'
        END
    ) || E'\n';

    v_result := v_result || format('  Throughput:   %s TPS → %s TPS    %s',
        round(v_data.tps_before, 0)::integer,
        round(v_data.tps_during, 0)::integer,
        CASE
            WHEN v_data.tps_change_pct > 0 THEN format('+%s%%', v_data.tps_change_pct::integer)
            WHEN v_data.tps_change_pct < 0 THEN format('%s%%', v_data.tps_change_pct::integer)
            ELSE 'unchanged'
        END
    ) || E'\n\n';

    -- =========================================================================
    -- Wait Events Section
    -- =========================================================================
    IF jsonb_array_length(v_data.top_wait_events) > 0 THEN
        v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';
        v_result := v_result || E'TOP WAIT EVENTS\n';
        v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';

        FOR v_wait IN SELECT * FROM jsonb_to_recordset(v_data.top_wait_events)
                      AS x(wait_type text, wait_event text, total_count integer, pct_increase numeric)
                      LIMIT 5 LOOP
            v_result := v_result || format('  %s:%s  count=%s',
                v_wait.wait_type, v_wait.wait_event, v_wait.total_count);
            IF v_wait.pct_increase IS NOT NULL THEN
                v_result := v_result || format('  (%s%s%%)',
                    CASE WHEN v_wait.pct_increase >= 0 THEN '+' ELSE '' END,
                    v_wait.pct_increase::integer);
            END IF;
            v_result := v_result || E'\n';
        END LOOP;
        v_result := v_result || E'\n';
    END IF;

    -- =========================================================================
    -- Recommendations Section
    -- =========================================================================
    v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';
    v_result := v_result || E'RECOMMENDATIONS\n';
    v_result := v_result || E'──────────────────────────────────────────────────────────────────────\n';

    FOR i IN 1..array_length(v_data.recommendations, 1) LOOP
        v_result := v_result || format('  • %s', v_data.recommendations[i]) || E'\n';
    END LOOP;
    v_result := v_result || E'\n';

    -- Footer
    v_result := v_result || E'══════════════════════════════════════════════════════════════════════\n';

    RETURN v_result;
END;
$$;
COMMENT ON FUNCTION pgfr_analyze.blast_radius_report IS
'Human-readable blast radius analysis report with ASCII-art formatting. Suitable for incident postmortems, Slack/email sharing, and documentation. Includes visual severity indicators, bar charts for lock types and affected apps, and actionable recommendations. Use: SELECT pgfr_analyze.blast_radius_report(''2024-01-15 10:23:00'', ''2024-01-15 10:35:00'');';

-- =============================================================================
-- V2 PARTITIONED TABLE READER FUNCTIONS (Issue #11)
-- Queries v2 tables (statement_snapshots_v2, table_snapshots_v2,
-- index_snapshots_v2) using int4 sample_ts ranges for partition pruning.
-- These are NEW functions — existing functions are untouched (backwards compat).
-- =============================================================================

-- Helper: convert timestamptz bounds to int4 sample_ts offsets
-- Used by all v2 reader functions to derive partition-prunable range predicates.

-- ==========================================================
-- pgfr_analyze/sql/10_v2_readers.sql
-- ==========================================================
-- SPDX-License-Identifier: Apache-2.0
-- Copyright 2026 David A. Ventimiglia

create or replace function pgfr_analyze.v2_time_range(
    p_start timestamptz,
    p_end   timestamptz
)
returns table (ts_start int4, ts_end int4)
language sql immutable as $$
    select
        extract(epoch from (p_start - pgfr_record.epoch()))::int4,
        extract(epoch from (p_end   - pgfr_record.epoch()))::int4
$$;
comment on function pgfr_analyze.v2_time_range(timestamptz, timestamptz) is
'Convert timestamptz bounds to int4 sample_ts offsets (seconds since pgfr_record.epoch()). '
'Used by v2 reader functions to produce partition-prunable range predicates on sample_ts.';


-- Returns top queries by total_exec_time delta over the specified time window,
-- reading directly from the v2 partitioned table using int4 sample_ts ranges.
-- JIT is disabled at entry to avoid planning regressions on partitioned tables.
create or replace function pgfr_analyze.statement_activity_v2(
    p_start timestamptz,
    p_end   timestamptz,
    p_limit integer default 25
)
returns table(
    queryid                  bigint,
    dbid                     oid,
    userid                   oid,
    toplevel                 boolean,
    calls_delta              bigint,
    total_exec_time_delta_ms float8,
    mean_exec_time_ms        float8,
    rows_delta               bigint,
    shared_blks_hit_delta    bigint,
    shared_blks_read_delta   bigint,
    temp_blks_written_delta  bigint,
    hit_ratio_pct            numeric,
    pgss_reset_warning       boolean  -- true if any row flagged a cluster-wide PGSS eviction/reset
)
language plpgsql volatile as $$
declare
    v_ts_start int4;
    v_ts_end   int4;
begin
    set local jit = off;

    select tr.ts_start, tr.ts_end
    into v_ts_start, v_ts_end
    from pgfr_analyze.v2_time_range(p_start, p_end) tr;

    return query
    with
    -- last known state BEFORE the window opens — the baseline for delta computation.
    -- Using sample_ts < v_ts_start (not >= v_ts_start) is critical: picking the first
    -- row inside the window produces deltas relative to an arbitrary mid-window
    -- snapshot, understating activity for queries that ran before the window opened.
    snap_start as (
        select distinct on (ss.queryid, ss.dbid, ss.userid, ss.toplevel)
            ss.queryid, ss.dbid, ss.userid, ss.toplevel,
            ss.calls, ss.total_exec_time, ss.rows,
            ss.shared_blks_hit, ss.shared_blks_read, ss.temp_blks_written
        from pgfr_record.statement_snapshots_v2 ss
        where ss.sample_ts < v_ts_start
        order by ss.queryid, ss.dbid, ss.userid, ss.toplevel, ss.sample_ts desc
    ),
    -- latest snapshot per query inside the window — the end of the delta
    snap_end as (
        select distinct on (se.queryid, se.dbid, se.userid, se.toplevel)
            se.queryid, se.dbid, se.userid, se.toplevel,
            se.calls, se.total_exec_time, se.mean_exec_time, se.rows,
            se.shared_blks_hit, se.shared_blks_read, se.temp_blks_written,
            se.pgss_dealloc_warning
        from pgfr_record.statement_snapshots_v2 se
        where se.sample_ts >= v_ts_start
          and se.sample_ts <  v_ts_end
        order by se.queryid, se.dbid, se.userid, se.toplevel, se.sample_ts desc
    )
    select
        e.queryid,
        e.dbid,
        e.userid,
        e.toplevel,
        greatest(0, e.calls - coalesce(s.calls, 0))             as calls_delta,
        greatest(0, e.total_exec_time - coalesce(s.total_exec_time, 0)) as total_exec_time_delta_ms,
        e.mean_exec_time                                         as mean_exec_time_ms,
        greatest(0, e.rows - coalesce(s.rows, 0))               as rows_delta,
        greatest(0, e.shared_blks_hit  - coalesce(s.shared_blks_hit, 0))  as shared_blks_hit_delta,
        greatest(0, e.shared_blks_read - coalesce(s.shared_blks_read, 0)) as shared_blks_read_delta,
        greatest(0, e.temp_blks_written - coalesce(s.temp_blks_written, 0)) as temp_blks_written_delta,
        case
            when (greatest(0, e.shared_blks_hit - coalesce(s.shared_blks_hit, 0))
                + greatest(0, e.shared_blks_read - coalesce(s.shared_blks_read, 0))) > 0
            then round(
                100.0 * greatest(0, e.shared_blks_hit - coalesce(s.shared_blks_hit, 0))::numeric
                / (greatest(0, e.shared_blks_hit - coalesce(s.shared_blks_hit, 0))
                 + greatest(0, e.shared_blks_read - coalesce(s.shared_blks_read, 0))),
                1
            )
            else null
        end as hit_ratio_pct,
        coalesce(e.pgss_dealloc_warning, false) as pgss_reset_warning
    from snap_end e
    left join snap_start s using (queryid, dbid, userid, toplevel)
    where greatest(0, e.total_exec_time - coalesce(s.total_exec_time, 0)) > 0
    order by total_exec_time_delta_ms desc
    limit p_limit;
end;
$$;
comment on function pgfr_analyze.statement_activity_v2(timestamptz, timestamptz, integer) is
'Return top queries by total_exec_time delta over a time window, reading from '
'statement_snapshots_v2 (v2 partitioned table). Uses int4 sample_ts for partition '
'pruning — no join to snapshots table. JIT disabled at entry. Backwards-compatible: '
'existing statement_compare() is untouched. '
'pgss_reset_warning = true means a cluster-wide PGSS eviction occurred at that tick — '
'deltas for that row may be understated due to counter reset.';


-- Returns tables ordered by modification rate (n_tup_ins + n_tup_upd + n_tup_del delta)
-- over the specified time window, reading from the v2 partitioned table.
-- JIT is disabled at entry to avoid planning regressions on partitioned tables.
create or replace function pgfr_analyze.table_activity_v2(
    p_start timestamptz,
    p_end   timestamptz,
    p_limit integer default 25
)
returns table(
    relid                   oid,
    dbid                    oid,
    n_tup_ins_delta         bigint,
    n_tup_upd_delta         bigint,
    n_tup_del_delta         bigint,
    n_tup_hot_upd_delta     bigint,
    seq_scan_delta          bigint,
    idx_scan_delta          bigint,
    total_modifications     bigint,
    n_live_tup              bigint,
    n_dead_tup              bigint,
    dead_tup_pct            numeric,
    table_size_bytes        bigint
)
language plpgsql volatile as $$
declare
    v_ts_start int4;
    v_ts_end   int4;
begin
    set local jit = off;

    select tr.ts_start, tr.ts_end
    into v_ts_start, v_ts_end
    from pgfr_analyze.v2_time_range(p_start, p_end) tr;

    return query
    with
    -- last known state before the window — baseline for delta computation
    snap_start as (
        select distinct on (ts.relid, ts.dbid)
            ts.relid, ts.dbid,
            ts.n_tup_ins, ts.n_tup_upd, ts.n_tup_del, ts.n_tup_hot_upd,
            ts.seq_scan, ts.idx_scan
        from pgfr_record.table_snapshots_v2 ts
        where ts.sample_ts < v_ts_start
        order by ts.relid, ts.dbid, ts.sample_ts desc
    ),
    -- latest snapshot inside the window — end of the delta
    snap_end as (
        select distinct on (te.relid, te.dbid)
            te.relid, te.dbid,
            te.n_tup_ins, te.n_tup_upd, te.n_tup_del, te.n_tup_hot_upd,
            te.seq_scan, te.idx_scan,
            te.n_live_tup, te.n_dead_tup,
            te.table_size_bytes
        from pgfr_record.table_snapshots_v2 te
        where te.sample_ts >= v_ts_start
          and te.sample_ts <  v_ts_end
        order by te.relid, te.dbid, te.sample_ts desc
    )
    select
        e.relid,
        e.dbid,
        greatest(0, e.n_tup_ins - coalesce(s.n_tup_ins, 0))     as n_tup_ins_delta,
        greatest(0, e.n_tup_upd - coalesce(s.n_tup_upd, 0))     as n_tup_upd_delta,
        greatest(0, e.n_tup_del - coalesce(s.n_tup_del, 0))     as n_tup_del_delta,
        greatest(0, e.n_tup_hot_upd - coalesce(s.n_tup_hot_upd, 0)) as n_tup_hot_upd_delta,
        greatest(0, e.seq_scan - coalesce(s.seq_scan, 0))        as seq_scan_delta,
        greatest(0, e.idx_scan - coalesce(s.idx_scan, 0))        as idx_scan_delta,
        greatest(0, e.n_tup_ins - coalesce(s.n_tup_ins, 0))
            + greatest(0, e.n_tup_upd - coalesce(s.n_tup_upd, 0))
            + greatest(0, e.n_tup_del - coalesce(s.n_tup_del, 0)) as total_modifications,
        e.n_live_tup,
        e.n_dead_tup,
        case
            when coalesce(e.n_live_tup, 0) + coalesce(e.n_dead_tup, 0) > 0
            then round(
                100.0 * coalesce(e.n_dead_tup, 0)::numeric
                / (coalesce(e.n_live_tup, 0) + coalesce(e.n_dead_tup, 0)),
                1
            )
            else 0::numeric
        end as dead_tup_pct,
        e.table_size_bytes
    from snap_end e
    left join snap_start s using (relid, dbid)
    order by total_modifications desc
    limit p_limit;
end;
$$;
comment on function pgfr_analyze.table_activity_v2(timestamptz, timestamptz, integer) is
'Return tables ordered by modification rate (ins+upd+del delta) over a time window, '
'reading from table_snapshots_v2 (v2 partitioned table). Uses int4 sample_ts for '
'partition pruning. JIT disabled at entry. Backwards-compatible: existing '
'table_compare() and table_hotspots() are untouched.';


-- Returns indexes ordered by idx_scan delta over the specified time window,
-- reading from the v2 partitioned table using int4 sample_ts ranges.
-- JIT is disabled at entry to avoid planning regressions on partitioned tables.
create or replace function pgfr_analyze.index_activity_v2(
    p_start timestamptz,
    p_end   timestamptz,
    p_limit integer default 25
)
returns table(
    relid              oid,
    indexrelid         oid,
    dbid               oid,
    idx_scan_delta     bigint,
    idx_tup_read_delta bigint,
    idx_tup_fetch_delta bigint,
    index_size_bytes   bigint,
    selectivity_pct    numeric
)
language plpgsql volatile as $$
declare
    v_ts_start int4;
    v_ts_end   int4;
begin
    set local jit = off;

    select tr.ts_start, tr.ts_end
    into v_ts_start, v_ts_end
    from pgfr_analyze.v2_time_range(p_start, p_end) tr;

    return query
    with
    -- last known state before the window — baseline for delta computation
    snap_start as (
        select distinct on (si.relid, si.indexrelid, si.dbid)
            si.relid, si.indexrelid, si.dbid,
            si.idx_scan, si.idx_tup_read, si.idx_tup_fetch
        from pgfr_record.index_snapshots_v2 si
        where si.sample_ts < v_ts_start
        order by si.relid, si.indexrelid, si.dbid, si.sample_ts desc
    ),
    -- latest snapshot inside the window — end of the delta
    snap_end as (
        select distinct on (ie.relid, ie.indexrelid, ie.dbid)
            ie.relid, ie.indexrelid, ie.dbid,
            ie.idx_scan, ie.idx_tup_read, ie.idx_tup_fetch,
            ie.index_size_bytes
        from pgfr_record.index_snapshots_v2 ie
        where ie.sample_ts >= v_ts_start
          and ie.sample_ts <  v_ts_end
        order by ie.relid, ie.indexrelid, ie.dbid, ie.sample_ts desc
    )
    select
        e.relid,
        e.indexrelid,
        e.dbid,
        greatest(0, e.idx_scan      - coalesce(s.idx_scan, 0))      as idx_scan_delta,
        greatest(0, e.idx_tup_read  - coalesce(s.idx_tup_read, 0))  as idx_tup_read_delta,
        greatest(0, e.idx_tup_fetch - coalesce(s.idx_tup_fetch, 0)) as idx_tup_fetch_delta,
        e.index_size_bytes,
        case
            when greatest(0, e.idx_tup_read - coalesce(s.idx_tup_read, 0)) > 0
            then round(
                100.0 * greatest(0, e.idx_tup_fetch - coalesce(s.idx_tup_fetch, 0))::numeric
                / greatest(0, e.idx_tup_read - coalesce(s.idx_tup_read, 0)),
                1
            )
            else null
        end as selectivity_pct
    from snap_end e
    left join snap_start s using (relid, indexrelid, dbid)
    order by idx_scan_delta desc
    limit p_limit;
end;
$$;
comment on function pgfr_analyze.index_activity_v2(timestamptz, timestamptz, integer) is
'Return indexes ordered by idx_scan delta over a time window, reading from '
'index_snapshots_v2 (v2 partitioned table). Uses int4 sample_ts for partition '
'pruning. JIT disabled at entry. Backwards-compatible: existing index_efficiency() '
'and unused_indexes() are untouched.';

--------------------------------------------------------------------------------
-- ring buffer v2 reader functions
-- replace old ring-table reads with wait_samples, activity_samples, lock_samples.
-- output signatures unchanged — callers keep working.
--------------------------------------------------------------------------------

-- recent_waits_current() v2
-- decodes wait_samples integer[] via wait_event_map.
-- retention window: ring_config.num_slots * rotation_period.
drop function if exists pgfr_analyze.recent_waits_current();
create or replace function pgfr_analyze.recent_waits_current()
returns table (
    captured_at     timestamptz,
    backend_state   text,
    wait_event_type text,
    wait_event      text,
    state           text,
    count           integer
)
language sql stable as $$
    with retention_cutoff as (
        select
            pgfr_record.epoch()
            + (
                extract(epoch from now() - pgfr_record.epoch())::int4
                - (num_slots * extract(epoch from rotation_period)::int4)
              ) * interval '1 second' as cutoff
        from pgfr_record.ring_config
        where singleton
    ),
    decoded as (
        select
            pgfr_record.epoch() + ws.sample_ts * interval '1 second' as captured_at,
            abs(ws.data[i])::smallint                                 as wait_id,
            ws.data[i + 1]::integer                                   as waiter_count
        from pgfr_record.wait_samples ws,
             retention_cutoff rc,
             generate_subscripts(ws.data, 1) as i
        where ws.data[i] < 0
          and (pgfr_record.epoch() + ws.sample_ts * interval '1 second') > rc.cutoff
    )
    select
        d.captured_at,
        wem.state        as backend_state,
        wem.type         as wait_event_type,
        wem.event        as wait_event,
        wem.state        as state,
        d.waiter_count   as count
    from decoded d
    join pgfr_record.wait_event_map wem on wem.id = d.wait_id
    order by d.captured_at desc, d.waiter_count desc;
$$;

comment on function pgfr_analyze.recent_waits_current() is
'Ring buffer v2: decode wait_samples integer[] via wait_event_map. '
'Returns same columns as the original recent_waits_current(). '
'Retention window: num_slots * rotation_period from ring_config.';

-- recent_activity_current() v2
-- reads activity_samples (flat rows, no decode needed).
create or replace function pgfr_analyze.recent_activity_current()
returns table (
    captured_at      timestamptz,
    pid              integer,
    usename          text,
    application_name text,
    backend_type     text,
    state            text,
    wait_event_type  text,
    wait_event       text,
    query_start      timestamptz,
    running_for      interval,
    query_preview    text
)
language sql stable as $$
    with retention_cutoff as (
        select
            pgfr_record.epoch()
            + (
                extract(epoch from now() - pgfr_record.epoch())::int4
                - (num_slots * extract(epoch from rotation_period)::int4)
              ) * interval '1 second' as cutoff
        from pgfr_record.ring_config
        where singleton
    )
    select
        pgfr_record.epoch() + as2.sample_ts * interval '1 second' as captured_at,
        as2.pid,
        as2.usename,
        as2.application_name,
        as2.backend_type,
        as2.state,
        as2.wait_event_type,
        as2.wait_event,
        as2.query_start,
        (pgfr_record.epoch() + as2.sample_ts * interval '1 second') - as2.query_start as running_for,
        as2.query_preview
    from pgfr_record.activity_samples as2,
         retention_cutoff rc
    where (pgfr_record.epoch() + as2.sample_ts * interval '1 second') > rc.cutoff
      and as2.pid is not null
    order by as2.sample_ts desc, as2.query_start asc;
$$;

comment on function pgfr_analyze.recent_activity_current() is
'Ring buffer v2: reads activity_samples (flat per-backend rows). '
'Returns same columns as the original recent_activity_current(). '
'Retention window: num_slots * rotation_period from ring_config.';

-- recent_locks_current() v2
-- reads lock_samples; decodes lock_type via lock_type_map.
-- blocked_user, blocked_app, query_preview: not stored in lock_samples v2
-- (lock_samples stores pids only) — returned as null for now.
create or replace function pgfr_analyze.recent_locks_current()
returns table (
    captured_at            timestamptz,
    blocked_pid            integer,
    blocked_user           text,
    blocked_app            text,
    blocked_duration       interval,
    blocking_pid           integer,
    blocking_user          text,
    blocking_app           text,
    lock_type              text,
    locked_relation        text,
    blocked_query_preview  text,
    blocking_query_preview text
)
language sql stable as $$
    with retention_cutoff as (
        select
            pgfr_record.epoch()
            + (
                extract(epoch from now() - pgfr_record.epoch())::int4
                - (num_slots * extract(epoch from rotation_period)::int4)
              ) * interval '1 second' as cutoff
        from pgfr_record.ring_config
        where singleton
    )
    select
        pgfr_record.epoch() + ls.sample_ts * interval '1 second'   as captured_at,
        ls.blocked_pid,
        null::text                                                  as blocked_user,
        null::text                                                  as blocked_app,
        ls.blocked_duration_s * interval '1 second'                as blocked_duration,
        ls.blocking_pid,
        null::text                                                  as blocking_user,
        null::text                                                  as blocking_app,
        coalesce(ltm.lock_type, ls.lock_type::text)                as lock_type,
        coalesce(
            (ls.locked_relation_oid::regclass)::text,
            'oid:' || ls.locked_relation_oid::text
        )                                                           as locked_relation,
        null::text                                                  as blocked_query_preview,
        null::text                                                  as blocking_query_preview
    from pgfr_record.lock_samples ls
    cross join retention_cutoff rc
    left join pgfr_record.lock_type_map ltm on ltm.id = ls.lock_type
    where (pgfr_record.epoch() + ls.sample_ts * interval '1 second') > rc.cutoff
    order by ls.sample_ts desc, ls.blocked_duration_s desc nulls last;
$$;

comment on function pgfr_analyze.recent_locks_current() is
'Ring buffer v2: reads lock_samples; decodes lock_type via lock_type_map. '
'blocked_user, blocked_app, query_preview are null in v2 (pids only stored). '
'Returns same columns as the original recent_locks_current().';

-- wait_summary() v2
-- decodes wait_samples integer[] for a given time window.
drop function if exists pgfr_analyze.wait_summary(timestamptz, timestamptz);
create or replace function pgfr_analyze.wait_summary(
    p_start_time timestamptz,
    p_end_time   timestamptz
)
returns table (
    backend_state   text,
    wait_event_type text,
    wait_event      text,
    sample_count    bigint,
    total_waiters   bigint,
    avg_waiters     numeric,
    max_waiters     integer,
    pct_of_samples  numeric
)
language sql stable as $$
    with bounds as (
        select
            extract(epoch from p_start_time - pgfr_record.epoch())::int4 as start_ts,
            extract(epoch from p_end_time   - pgfr_record.epoch())::int4 as end_ts
    ),
    in_range as (
        select
            ws.sample_ts,
            abs(ws.data[i])::smallint  as wait_id,
            ws.data[i + 1]::integer    as waiter_count
        from pgfr_record.wait_samples ws,
             bounds b,
             generate_subscripts(ws.data, 1) as i
        where ws.data[i] < 0
          and ws.sample_ts between b.start_ts and b.end_ts
    ),
    total_samples as (
        select count(distinct sample_ts) as cnt from in_range
    ),
    grouped as (
        select
            wait_id,
            count(distinct sample_ts)   as sample_count,
            sum(waiter_count)           as total_waiters,
            round(avg(waiter_count), 2) as avg_waiters,
            max(waiter_count)           as max_waiters
        from in_range
        group by wait_id
    )
    select
        wem.state        as backend_state,
        wem.type         as wait_event_type,
        wem.event        as wait_event,
        g.sample_count,
        g.total_waiters,
        g.avg_waiters,
        g.max_waiters::integer,
        round(100.0 * g.sample_count / nullif(t.cnt, 0), 1) as pct_of_samples
    from grouped g
    cross join total_samples t
    join pgfr_record.wait_event_map wem on wem.id = g.wait_id
    order by g.total_waiters desc, g.sample_count desc;
$$;

comment on function pgfr_analyze.wait_summary(timestamptz, timestamptz) is
'Ring buffer v2: decode wait_samples integer[] for a time window. '
'Returns same columns as original wait_summary(). '
'Uses int4 sample_ts bounds for partition pruning.';


COMMIT;
