-- ============================================================================
-- lock_monitoring.sql
--
-- Always-on, pooler-proof monitoring for the Supavisor connection-spike /
-- advisory-lock-contention incidents on the shared (schema-isolated) Honcho DB.
--
-- WHY pg_cron: during a spike, Supavisor is saturated and you CANNOT get a
-- client connection to run ad-hoc diagnostics. pg_cron jobs run in background
-- workers INSIDE Postgres -- they don't go through Supavisor -- so they keep
-- recording straight through a 9,000-client collapse. The server side (~155
-- backends) is never the thing that saturates, so there's always room to run.
--
-- WHAT IT CANNOT SEE: pg_stat_activity only shows the ~155 server backends.
-- The 9,000 figure is CLIENT connections queued on Supavisor that never got a
-- backend -- they have no query to show. The intended query of a client stuck
-- waiting for a slot needs app-side instrumentation (Honcho sets
-- application_name to the route/context; plus Sentry/Prometheus). What this
-- captures is the CAUSE: which backends are stuck, on what, and whether the
-- lock table is filling -- the early-warning signal that precedes the avalanche.
--
-- Objects live in the public schema (empty on tenant DBs; Honcho tables live in
-- the per-tenant DB.SCHEMA). All prefixed mon_ so they stay grouped.
--
-- Layout:
--   SECTION 1  Install (tables + sampler functions)
--   SECTION 2  Schedule (pg_cron)
--   SECTION 3  Dry-run / verify
--   SECTION 4  Diagnostic runbook (Q0-Q6) -- run AFTER a spike
--   SECTION 5  Teardown
--   SECTION 6  Recommended server-side settings (separate from this script)
-- ============================================================================


-- ============================================================================
-- SECTION 1 -- INSTALL
-- ============================================================================

-- A) Connection-state breakdown -- the "is it tipping?" timeline (cheap, low volume)
CREATE TABLE IF NOT EXISTS public.mon_conn_samples (
  sampled_at      timestamptz NOT NULL DEFAULT clock_timestamp(),
  state           text,
  wait_event_type text,
  wait_event      text,
  backend_count   int
);
CREATE INDEX IF NOT EXISTS mon_conn_samples_at ON public.mon_conn_samples (sampled_at);

-- B) Detailed waiters/blockers -- only non-idle or waiting backends (keeps volume sane)
CREATE TABLE IF NOT EXISTS public.mon_activity_samples (
  sampled_at       timestamptz NOT NULL DEFAULT clock_timestamp(),
  pid              int,
  state            text,
  wait_event_type  text,
  wait_event       text,
  xact_age         interval,
  query_age        interval,
  blocked_by       int[],
  application_name text,   -- HonchoAsyncSession sets this to the route/context
  datname          text,
  usename          text,
  query            text
);
CREATE INDEX IF NOT EXISTS mon_activity_samples_at ON public.mon_activity_samples (sampled_at);

-- C) Lock-table pressure -- catches the "out of shared memory" precursor
CREATE TABLE IF NOT EXISTS public.mon_lock_samples (
  sampled_at  timestamptz NOT NULL DEFAULT clock_timestamp(),
  locktype    text,
  mode        text,
  granted     bool,
  lock_count  int
);
CREATE INDEX IF NOT EXISTS mon_lock_samples_at ON public.mon_lock_samples (sampled_at);

-- D) pg_stat_statements snapshots -- query-level cost attribution (cumulative; diff across window)
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;   -- already preloaded on Supabase
CREATE TABLE IF NOT EXISTS public.mon_stmt_samples (
  sampled_at      timestamptz NOT NULL DEFAULT clock_timestamp(),
  queryid         bigint,
  calls           bigint,
  total_exec_time double precision,
  rows            bigint,
  query           text
);
CREATE INDEX IF NOT EXISTS mon_stmt_samples_at ON public.mon_stmt_samples (sampled_at);

-- Sampler: one call captures conn + activity + lock layers.
CREATE OR REPLACE FUNCTION public.mon_sample_db_state() RETURNS void
LANGUAGE plpgsql AS $fn$
BEGIN
  INSERT INTO public.mon_conn_samples (state, wait_event_type, wait_event, backend_count)
  SELECT state, wait_event_type, wait_event, count(*)
  FROM pg_stat_activity
  WHERE backend_type = 'client backend'
  GROUP BY 1, 2, 3;

  INSERT INTO public.mon_activity_samples
    (pid, state, wait_event_type, wait_event, xact_age, query_age,
     blocked_by, application_name, datname, usename, query)
  SELECT pid, state, wait_event_type, wait_event,
         clock_timestamp() - xact_start, clock_timestamp() - query_start,
         pg_blocking_pids(pid), application_name, datname, usename, left(query, 500)
  FROM pg_stat_activity
  WHERE backend_type = 'client backend'
    AND pid <> pg_backend_pid()
    AND (state IS DISTINCT FROM 'idle' OR wait_event IS NOT NULL);

  INSERT INTO public.mon_lock_samples (locktype, mode, granted, lock_count)
  SELECT locktype, mode, granted, count(*)
  FROM pg_locks
  GROUP BY 1, 2, 3;
END;
$fn$;

-- Sampler: pg_stat_statements top-100 by cumulative cost (normalized, so same set across tenants).
CREATE OR REPLACE FUNCTION public.mon_sample_statements() RETURNS void
LANGUAGE plpgsql AS $fn$
BEGIN
  INSERT INTO public.mon_stmt_samples (queryid, calls, total_exec_time, rows, query)
  SELECT queryid, calls, total_exec_time, rows, left(query, 300)
  FROM pg_stat_statements
  ORDER BY total_exec_time DESC
  LIMIT 100;
END;
$fn$;


-- ============================================================================
-- SECTION 2 -- SCHEDULE (pg_cron)
--
-- Interval rule: '[1-59] seconds' for sub-minute; standard 5-field cron
-- ('* * * * *', '*/10 * * * *') for >= 1 minute. ('60 seconds' is INVALID.)
--
-- Prefer the '5 seconds' schedule (short independent transactions) over a
-- self-looping minute job -- a loop with pg_sleep holds one transaction open
-- for ~56s/min, pinning the xmin horizon and holding back vacuum.
-- ============================================================================

SELECT cron.schedule('mon-db-state-sampler', '5 seconds',
  $$SELECT public.mon_sample_db_state();$$);

SELECT cron.schedule('mon-stmt-sampler', '* * * * *',
  $$SELECT public.mon_sample_statements();$$);

SELECT cron.schedule('mon-db-state-retention', '*/10 * * * *', $$
  DELETE FROM public.mon_conn_samples     WHERE sampled_at < now() - interval '7 days';
  DELETE FROM public.mon_activity_samples WHERE sampled_at < now() - interval '7 days';
  DELETE FROM public.mon_lock_samples     WHERE sampled_at < now() - interval '7 days';
  DELETE FROM public.mon_stmt_samples     WHERE sampled_at < now() - interval '7 days';
$$);


-- ============================================================================
-- SECTION 3 -- DRY-RUN / VERIFY
-- ============================================================================

-- Run the sampler once by hand and eyeball volume before trusting the schedule:
--   SELECT public.mon_sample_db_state();
--   SELECT count(*) FROM public.mon_activity_samples;   -- waiters right now
--   SELECT count(*) FROM public.mon_lock_samples;       -- distinct lock groups
--   TABLE public.mon_conn_samples;                       -- state breakdown

-- Confirm jobs are registered + succeeding:
--   SELECT jobid, jobname, schedule, active FROM cron.job WHERE jobname LIKE 'mon-%';
--   SELECT jobname, status, start_time FROM cron.job_run_details
--   WHERE jobname LIKE 'mon-%' ORDER BY start_time DESC LIMIT 10;

-- Footprint check anytime:
--   SELECT relname, pg_size_pretty(pg_total_relation_size('public.'||relname)) AS size
--   FROM (VALUES ('mon_conn_samples'),('mon_activity_samples'),
--                ('mon_lock_samples'),('mon_stmt_samples')) v(relname);


-- ============================================================================
-- SECTION 4 -- DIAGNOSTIC RUNBOOK (run AFTER a spike, against recorded history)
--
-- Q0 finds the window. Then edit w.lo / w.hi in Q1-Q6 to bracket it.
-- Decision chain: Q1 = WHEN it tipped -> Q2/Q5 = WHICH statement was at the
-- head -> Q3/Q4 = WHICH mechanism (lock-table exhaustion vs advisory pileup)
-- -> Q6 = aggregate query load corroboration.
-- ============================================================================

-- ---- Q0 -- Find the spike window automatically (worst minutes first) ----
SELECT date_trunc('minute', sampled_at) AS minute,
       max(s.total)        AS peak_backends,
       max(s.lock_waiters) AS peak_lock_waiters,
       max(s.idle_in_txn)  AS peak_idle_in_txn,
       max(s.active)       AS peak_active
FROM (
  SELECT sampled_at,
         sum(backend_count)                                              AS total,
         sum(backend_count) FILTER (WHERE wait_event_type = 'Lock')      AS lock_waiters,
         sum(backend_count) FILTER (WHERE state = 'idle in transaction') AS idle_in_txn,
         sum(backend_count) FILTER (WHERE state = 'active')              AS active
  FROM public.mon_conn_samples
  GROUP BY sampled_at
) s
GROUP BY 1
ORDER BY peak_backends DESC
LIMIT 30;


-- ---- Q1 -- Timeline: backend state/wait through the spike ----
WITH w AS (SELECT '2026-06-01 23:00-04'::timestamptz lo, '2026-06-01 23:30-04'::timestamptz hi)
SELECT date_trunc('second', sampled_at) AS t,
       sum(backend_count)                                              AS total,
       sum(backend_count) FILTER (WHERE state = 'active')              AS active,
       sum(backend_count) FILTER (WHERE state = 'idle in transaction') AS idle_in_txn,
       sum(backend_count) FILTER (WHERE wait_event_type = 'Lock')      AS lock_waiters,
       sum(backend_count) FILTER (WHERE wait_event = 'advisory')       AS advisory_waiters
FROM public.mon_conn_samples, w
WHERE sampled_at BETWEEN w.lo AND w.hi
GROUP BY 1 ORDER BY 1;


-- ---- Q2 -- What were the stuck backends running? (smoking-gun statements) ----
-- Top row 'Lock/advisory' on INSERT...messages -> advisory-lock trigger (DEV-1852).
-- Top row 'idle in transaction' w/ large longest_txn -> leaked transaction.
-- Top row 'active' on one heavy query -> a slow query is the stall point.
WITH w AS (SELECT '2026-06-01 23:00-04'::timestamptz lo, '2026-06-01 23:30-04'::timestamptz hi)
SELECT wait_event_type, wait_event, state, application_name,
       left(query, 100) AS query,
       count(*)         AS samples,
       max(xact_age)    AS longest_txn
FROM public.mon_activity_samples, w
WHERE sampled_at BETWEEN w.lo AND w.hi
GROUP BY 1,2,3,4,5
ORDER BY samples DESC
LIMIT 30;


-- ---- Q3 -- Was the lock table filling? ("out of shared memory" test) ----
WITH w AS (SELECT '2026-06-01 23:00-04'::timestamptz lo, '2026-06-01 23:30-04'::timestamptz hi)
SELECT date_trunc('second', sampled_at) AS t, sum(lock_count) AS total_locks
FROM public.mon_lock_samples, w
WHERE sampled_at BETWEEN w.lo AND w.hi
GROUP BY 1 ORDER BY total_locks DESC LIMIT 20;

-- The hard ceiling to compare total_locks against:
SELECT current_setting('max_locks_per_transaction')::int
     * (current_setting('max_connections')::int
        + current_setting('max_prepared_transactions')::int) AS lock_table_slots;


-- ---- Q4 -- Advisory locks held vs waiting over time ----
WITH w AS (SELECT '2026-06-01 23:00-04'::timestamptz lo, '2026-06-01 23:30-04'::timestamptz hi)
SELECT date_trunc('second', sampled_at) AS t,
       sum(lock_count) FILTER (WHERE granted)     AS held,
       sum(lock_count) FILTER (WHERE NOT granted) AS waiting
FROM public.mon_lock_samples, w
WHERE locktype = 'advisory' AND sampled_at BETWEEN w.lo AND w.hi
GROUP BY 1 HAVING sum(lock_count) FILTER (WHERE NOT granted) > 0
ORDER BY waiting DESC;


-- ---- Q5 -- Blocking chains: who blocked whom ----
-- The most frequent blocker_pid/blocker_query is the head of the pileup.
-- (Holders are always in a transaction, so captured; null blocker_query just
--  means the holder was plain-idle at that instant.)
WITH w AS (SELECT '2026-06-01 23:00-04'::timestamptz lo, '2026-06-01 23:30-04'::timestamptz hi)
SELECT a.sampled_at,
       a.pid              AS waiter_pid,
       a.wait_event,
       left(a.query, 70)  AS waiter_query,
       bp.b_pid           AS blocker_pid,
       blk.state          AS blocker_state,
       left(blk.query,70) AS blocker_query
FROM public.mon_activity_samples a, w
CROSS JOIN LATERAL unnest(a.blocked_by) AS bp(b_pid)
LEFT JOIN public.mon_activity_samples blk
       ON blk.sampled_at = a.sampled_at AND blk.pid = bp.b_pid
WHERE a.sampled_at BETWEEN w.lo AND w.hi
  AND cardinality(a.blocked_by) > 0
ORDER BY a.sampled_at, waiter_pid
LIMIT 200;


-- ---- Q6 -- Which queries ran the most during the window (pg_stat_statements delta) ----
WITH w AS (SELECT '2026-06-01 23:00-04'::timestamptz lo, '2026-06-01 23:30-04'::timestamptz hi),
b AS (SELECT min(sampled_at) lo, max(sampled_at) hi FROM public.mon_stmt_samples, w
      WHERE sampled_at BETWEEN w.lo AND w.hi),
first AS (SELECT DISTINCT ON (queryid) queryid, calls, total_exec_time, query
          FROM public.mon_stmt_samples, b WHERE sampled_at = b.lo ORDER BY queryid),
last  AS (SELECT DISTINCT ON (queryid) queryid, calls, total_exec_time
          FROM public.mon_stmt_samples, b WHERE sampled_at = b.hi ORDER BY queryid)
SELECT f.query,
       l.calls - f.calls AS calls_in_window,
       round((l.total_exec_time - f.total_exec_time)::numeric, 1) AS exec_ms_in_window
FROM last l JOIN first f USING (queryid)
WHERE l.calls > f.calls
ORDER BY exec_ms_in_window DESC
LIMIT 25;


-- ============================================================================
-- SECTION 5 -- TEARDOWN
-- ============================================================================
-- SELECT cron.unschedule('mon-db-state-sampler');
-- SELECT cron.unschedule('mon-stmt-sampler');
-- SELECT cron.unschedule('mon-db-state-retention');
-- DROP FUNCTION IF EXISTS public.mon_sample_db_state();
-- DROP FUNCTION IF EXISTS public.mon_sample_statements();
-- DROP TABLE IF EXISTS public.mon_conn_samples, public.mon_activity_samples,
--                      public.mon_lock_samples, public.mon_stmt_samples;


-- ============================================================================
-- SECTION 6 -- RECOMMENDED SERVER-SIDE SETTINGS (not part of this script)
--
-- Ask Supabase to apply; all but max_locks_per_transaction are SIGHUP (no restart).
--   ALTER SYSTEM SET log_lock_waits = on;            -- log every lock wait + blocker
--   ALTER SYSTEM SET deadlock_timeout = '1s';        -- also the lock-wait log threshold
--   ALTER SYSTEM SET log_min_duration_statement = '500ms';
--   ALTER SYSTEM SET max_locks_per_transaction = 256; -- from 64; raises tipping point (RESTART)
--   SELECT pg_reload_conf();
-- These feed the Postgres logs you already ship to Logflare -- full statement
-- text + lock-wait detail, including the contended advisory lock.
-- ============================================================================
