-- Connect as superuser
CREATE EXTENSION pg_background;

-- Verify installation
SELECT extname, extversion FROM pg_extension WHERE extname = 'pg_background';
-- Expected output:
--    extname     | extversion 
-- ---------------+------------
--  pg_background | 1.6

-- Set worker process limit (adjust based on your workload)
ALTER SYSTEM SET max_worker_processes = 32;

-- Reload configuration
SELECT pg_reload_conf();

-- Verify setting
SHOW max_worker_processes;

-- Launch worker and capture handle
SELECT * FROM pg_background_launch_v2(
  'SELECT pg_sleep(5); SELECT count(*) FROM large_table'
) AS handle;

-- Output:
--   pid  |      cookie       
-- -------+-------------------
--  12345 | 1234567890123456

-- Results can only be consumed ONCE
SELECT * FROM pg_background_result_v2(12345, 1234567890123456) AS (count BIGINT);

-- Attempting second retrieval will error:
-- ERROR: results already consumed for worker PID 12345

-- For queries with side effects only (no result consumption needed)
SELECT * FROM pg_background_submit_v2(
  'INSERT INTO audit_log (ts, event) VALUES (now(), ''system_check'')'
) AS handle;

-- Worker commits and exits automatically

-- Request immediate cancellation
SELECT pg_background_cancel_v2(pid, cookie);

-- Or with grace period (500ms to finish current statement)
SELECT pg_background_cancel_v2_grace(pid, cookie, 500);

-- Block until worker finishes
SELECT pg_background_wait_v2(pid, cookie);

-- Or wait with timeout (returns true if completed)
SELECT pg_background_wait_v2_timeout(pid, cookie, 5000);  -- 5 seconds

SELECT *
FROM pg_background_list_v2()
AS (
  pid int4,
  cookie int8,
  launched_at timestamptz,
  user_id oid,
  queue_size int4,
  state text,
  sql_preview text,
  last_error text,
  consumed bool
)
ORDER BY launched_at DESC;

-- Launch (returns bare PID)
SELECT pg_background_launch('VACUUM VERBOSE my_table') AS pid \gset

-- Retrieve results
SELECT * FROM pg_background_result(:pid) AS (result TEXT);

-- Fire-and-forget (detach does NOT cancel!)
SELECT pg_background_detach(:pid);

CREATE TYPE public.pg_background_handle AS (
  pid    int4,   -- Process ID
  cookie int8    -- Unique identifier (prevents PID reuse)
);

-- Launch worker that sends notification
SELECT * FROM pg_background_launch_v2(
  $$SELECT pg_notify('alerts', 'system_event')$$
) AS h \gset

-- Detach only removes launcher's tracking
SELECT pg_background_detach_v2(:'h.pid', :'h.cookie');

-- Worker STILL runs and sends notification!
-- To actually prevent notification, use:
SELECT pg_background_cancel_v2(:'h.pid', :'h.cookie');

-- Day 1: Launch worker
SELECT pg_background_launch('slow_query()') AS pid \gset

-- Day 2: Session still alive, but worker PID may be reused
-- This could attach to a DIFFERENT worker with the SAME PID!
SELECT pg_background_result(:pid);  -- ⚠️ DANGEROUS

-- Launch with cookie
SELECT * FROM pg_background_launch_v2('slow_query()') AS h \gset

-- Days later: cookie validation prevents mismatch
SELECT pg_background_result_v2(:'h.pid', :'h.cookie');
-- If PID reused, cookie won't match → safe error

BEGIN;
  -- Launcher transaction starts

  SELECT * FROM pg_background_launch_v2(
    'INSERT INTO audit_log VALUES (now(), ''user_action'')'
  ) AS h \gset;
  
  -- Main work
  UPDATE users SET status = 'active' WHERE id = 123;
  
  -- If we ROLLBACK, the audit_log INSERT still commits!
ROLLBACK;

-- audit_log entry exists despite rollback

-- Worker sends notification
SELECT * FROM pg_background_launch_v2(
  $$SELECT pg_notify('channel', 'message')$$
) AS h \gset;

-- Detach removes tracking but does NOT cancel
SELECT pg_background_detach_v2(:'h.pid', :'h.cookie');

-- Notification WILL be delivered (worker commits independently)

-- Cancel before worker commits
SELECT pg_background_cancel_v2(:'h.pid', :'h.cookie');

-- Extension creates this role automatically:
CREATE ROLE pgbackground_role NOLOGIN INHERIT;

-- All pg_background functions granted to this role
-- PUBLIC has NO access by default

-- Method 1: Direct role grant (recommended)
GRANT pgbackground_role TO app_user;

-- Method 2: Helper function (explicit EXECUTE grants)
SELECT grant_pg_background_privileges('app_user', true);

-- Method 1: Revoke role membership
REVOKE pgbackground_role FROM app_user;

-- Method 2: Helper function
SELECT revoke_pg_background_privileges('app_user', true);

CREATE FUNCTION unsafe_launch(user_input text) RETURNS void AS $$
BEGIN
  -- NEVER concatenate untrusted input!
  PERFORM pg_background_launch_v2(
    'SELECT * FROM users WHERE name = ''' || user_input || ''''
  );
END;
$$ LANGUAGE plpgsql;

CREATE FUNCTION safe_launch(user_input text) RETURNS void AS $$
BEGIN
  -- Use %L for literal quoting
  PERFORM pg_background_launch_v2(
    format('SELECT * FROM users WHERE name = %L', user_input)
  );
END;
$$ LANGUAGE plpgsql;

-- Application-level quota enforcement
CREATE OR REPLACE FUNCTION launch_with_limit(sql text)
RETURNS pg_background_handle AS $$
DECLARE
  active_count int;
  h pg_background_handle;
BEGIN
  -- Count active workers for current user
  SELECT count(*) INTO active_count
  FROM pg_background_list_v2() AS (
    pid int4, cookie int8, launched_at timestamptz, user_id oid,
    queue_size int4, state text, sql_preview text, last_error text, consumed bool
  )
  WHERE user_id = current_user::regrole::oid
    AND state IN ('running');
  
  IF active_count >= 5 THEN
    RAISE EXCEPTION 'User worker limit exceeded (max 5 concurrent)';
  END IF;
  
  SELECT * INTO h FROM pg_background_launch_v2(sql);
  RETURN h;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- list_v2() exposes SQL previews (first 120 chars) and error messages
-- For sensitive deployments, create restricted view:

CREATE VIEW public.safe_worker_list AS
SELECT pid, cookie, state, consumed, launched_at
FROM pg_background_list_v2() AS (
  pid int4, cookie int8, launched_at timestamptz, user_id oid,
  queue_size int4, state text, sql_preview text, last_error text, consumed bool
)
WHERE user_id = current_user::regrole::oid;
-- Omit sql_preview and last_error

GRANT SELECT ON public.safe_worker_list TO app_users;

-- Launch background VACUUM
SELECT * FROM pg_background_launch_v2(
  'VACUUM (VERBOSE, ANALYZE) large_table'
) AS h \gset

-- Check progress periodically
SELECT state, sql_preview
FROM pg_background_list_v2() AS (
  pid int4, cookie int8, launched_at timestamptz, user_id oid,
  queue_size int4, state text, sql_preview text, last_error text, consumed bool
)
WHERE pid = :'h.pid' AND cookie = :'h.cookie';

-- Wait for completion (optional)
SELECT pg_background_wait_v2(:'h.pid', :'h.cookie');

-- Cleanup tracking
SELECT pg_background_detach_v2(:'h.pid', :'h.cookie');

CREATE FUNCTION log_audit(event_type text, details jsonb)
RETURNS void AS $$
DECLARE
  h pg_background_handle;
BEGIN
  -- Launch audit insert (commits independently)
  SELECT * INTO h FROM pg_background_submit_v2(
    format(
      'INSERT INTO audit_log (ts, event_type, details) VALUES (now(), %L, %L)',
      event_type,
      details::text
    )
  );
  
  -- Detach immediately (fire-and-forget)
  PERFORM pg_background_detach_v2(h.pid, h.cookie);
END;
$$ LANGUAGE plpgsql;

-- Usage in transaction
BEGIN;
  UPDATE accounts SET balance = balance - 100 WHERE id = 123;
  
  -- Audit log commits even if UPDATE rolls back
  PERFORM log_audit('withdrawal', '{"account": 123, "amount": 100}');
  
  -- Simulate error
  ROLLBACK;
  
-- Audit log entry exists!
SELECT * FROM audit_log ORDER BY ts DESC LIMIT 1;

CREATE FUNCTION notify_async(channel text, payload text)
RETURNS void AS $$
DECLARE
  h pg_background_handle;
BEGIN
  SELECT * INTO h FROM pg_background_submit_v2(
    format('SELECT pg_notify(%L, %L)', channel, payload)
  );
  
  PERFORM pg_background_detach_v2(h.pid, h.cookie);
END;
$$ LANGUAGE plpgsql;

-- Usage
SELECT notify_async('order_updates', '{"order_id": 456, "status": "shipped"}');

-- Launch ETL
SELECT * FROM pg_background_launch_v2($$
  INSERT INTO fact_sales
  SELECT * FROM staging_sales
  WHERE processed = false;
  
  UPDATE staging_sales SET processed = true;
$$) AS h \gset

-- Store handle for later retrieval
INSERT INTO job_tracker (job_id, pid, cookie, started_at)
VALUES ('etl-001', :'h.pid', :'h.cookie', now());

-- Later: check status
SELECT
  j.job_id,
  w.state,
  w.launched_at,
  (now() - w.launched_at) AS duration
FROM job_tracker j
CROSS JOIN LATERAL (
  SELECT *
  FROM pg_background_list_v2() AS (
    pid int4, cookie int8, launched_at timestamptz, user_id oid,
    queue_size int4, state text, sql_preview text, last_error text, consumed bool
  )
  WHERE pid = j.pid AND cookie = j.cookie
) w
WHERE j.job_id = 'etl-001';

DO $$
DECLARE
  h1 pg_background_handle;
  h2 pg_background_handle;
  h3 pg_background_handle;
  total_rows bigint;
BEGIN
  -- Launch parallel workers
  SELECT * INTO h1 FROM pg_background_launch_v2('SELECT count(*) FROM sales');
  SELECT * INTO h2 FROM pg_background_launch_v2('SELECT count(*) FROM orders');
  SELECT * INTO h3 FROM pg_background_launch_v2('SELECT count(*) FROM customers');
  
  -- Wait for all to complete
  PERFORM pg_background_wait_v2(h1.pid, h1.cookie);
  PERFORM pg_background_wait_v2(h2.pid, h2.cookie);
  PERFORM pg_background_wait_v2(h3.pid, h3.cookie);
  
  -- Aggregate results
  SELECT sum(cnt) INTO total_rows FROM (
    SELECT * FROM pg_background_result_v2(h1.pid, h1.cookie) AS (cnt bigint)
    UNION ALL
    SELECT * FROM pg_background_result_v2(h2.pid, h2.cookie) AS (cnt bigint)
    UNION ALL
    SELECT * FROM pg_background_result_v2(h3.pid, h3.cookie) AS (cnt bigint)
  ) t;
  
  RAISE NOTICE 'Total rows: %', total_rows;
END;
$$;

CREATE FUNCTION run_with_timeout(sql text, timeout_sec int)
RETURNS text AS $$
DECLARE
  h pg_background_handle;
  done bool;
  result_text text;
BEGIN
  -- Launch worker
  SELECT * INTO h FROM pg_background_launch_v2(sql);
  
  -- Wait with timeout
  done := pg_background_wait_v2_timeout(h.pid, h.cookie, timeout_sec * 1000);
  
  IF NOT done THEN
    -- Timeout exceeded
    RAISE WARNING 'Query timed out after % seconds, cancelling...', timeout_sec;
    PERFORM pg_background_cancel_v2_grace(h.pid, h.cookie, 1000);
    PERFORM pg_background_detach_v2(h.pid, h.cookie);
    RETURN 'TIMEOUT';
  END IF;
  
  -- Retrieve result
  SELECT * INTO result_text FROM pg_background_result_v2(h.pid, h.cookie) AS (res text);
  RETURN result_text;
END;
$$ LANGUAGE plpgsql;

-- Usage
SELECT run_with_timeout('SELECT pg_sleep(10)', 5);  -- Returns 'TIMEOUT'

SELECT count(*) AS bgworker_count
FROM pg_stat_activity
WHERE backend_type LIKE '%background%';

-- Formula: autovacuum_workers + max_parallel_workers + pg_background_estimate + buffer
ALTER SYSTEM SET max_worker_processes = 64;  -- Adjust per workload
SELECT pg_reload_conf();

SELECT
  name,
  size,
  allocated_size
FROM pg_shmem_allocations
WHERE name LIKE '%pg_background%'
ORDER BY size DESC;

-- Small results (default)
SELECT pg_background_launch_v2('SELECT id FROM small_table', 65536);

-- Large results (1MB queue)
SELECT pg_background_launch_v2('SELECT * FROM huge_table', 1048576);

-- Detach all completed workers
DO $$
DECLARE r record;
BEGIN
  FOR r IN
    SELECT *
    FROM pg_background_list_v2() AS (
      pid int4, cookie int8, launched_at timestamptz, user_id oid,
      queue_size int4, state text, sql_preview text, last_error text, consumed bool
    )
    WHERE state IN ('stopped', 'canceled', 'error')
  LOOP
    PERFORM pg_background_detach_v2(r.pid, r.cookie);
  END LOOP;
END;
$$;

-- Check if worker is blocking on queue send
SELECT
  pid,
  state,
  wait_event_type,
  wait_event
FROM pg_stat_activity
WHERE backend_type LIKE '%background%'
  AND wait_event = 'SHM_MQ_SEND';

-- Temporarily increase timeout
SET statement_timeout = '30min';
SELECT pg_background_launch_v2('slow_aggregation_query()');
RESET statement_timeout;

ALTER DATABASE production SET statement_timeout = '10min';

-- Include SET in worker SQL
SELECT pg_background_launch_v2($$
  SET work_mem = '256MB';
  SELECT * FROM large_table ORDER BY col;
$$);

-- Both settings are independent
ALTER SYSTEM SET max_worker_processes = 64;     -- Total pool
ALTER SYSTEM SET max_parallel_workers = 16;     -- Parallel query subset

CREATE VIEW pg_background_status AS
SELECT
  w.pid,
  w.cookie,
  w.state,
  left(w.sql_preview, 60) AS sql_snippet,
  w.launched_at,
  (now() - w.launched_at) AS age,
  w.consumed,
  a.state AS pg_state,
  a.wait_event_type,
  a.wait_event,
  a.query AS current_query
FROM pg_background_list_v2() AS (
  pid int4, cookie int8, launched_at timestamptz, user_id oid,
  queue_size int4, state text, sql_preview text, last_error text, consumed bool
) w
LEFT JOIN pg_stat_activity a USING (pid)
ORDER BY w.launched_at DESC;

-- Query it
SELECT * FROM pg_background_status;

-- Workers running > 1 hour
SELECT
  pid,
  cookie,
  sql_preview,
  (now() - launched_at) AS duration
FROM pg_background_list_v2() AS (
  pid int4, cookie int8, launched_at timestamptz, user_id oid,
  queue_size int4, state text, sql_preview text, last_error text, consumed bool
)
WHERE state = 'running'
  AND (now() - launched_at) > interval '1 hour';

-- Export metrics for monitoring systems
SELECT
  'pg_background_active_workers' AS metric,
  count(*) AS value,
  state AS labels
FROM pg_background_list_v2() AS (
  pid int4, cookie int8, launched_at timestamptz, user_id oid,
  queue_size int4, state text, sql_preview text, last_error text, consumed bool
)
GROUP BY state;

-- Check current limit and usage
SHOW max_worker_processes;
SELECT count(*) FROM pg_stat_activity WHERE backend_type LIKE '%worker%';

-- Increase limit (requires restart for some versions)
ALTER SYSTEM SET max_worker_processes = 32;
SELECT pg_reload_conf();  -- Or restart PostgreSQL

-- ❌ Bad: Reusing old handle
-- h was from hours ago, worker exited, PID reused

-- ✅ Good: Fresh handle per operation
SELECT * FROM pg_background_launch_v2('...') AS h \gset
SELECT pg_background_wait_v2(:'h.pid', :'h.cookie');

-- Check what worker is waiting on
SELECT
  w.pid,
  w.sql_preview,
  a.wait_event_type,
  a.wait_event,
  a.state,
  a.query
FROM pg_background_list_v2() AS (
  pid int4, cookie int8, launched_at timestamptz, user_id oid,
  queue_size int4, state text, sql_preview text, last_error text, consumed bool
) w
JOIN pg_stat_activity a USING (pid)
WHERE w.state = 'running';

-- Check locks
SELECT
  l.pid,
  l.locktype,
  l.relation::regclass,
  l.mode,
  l.granted
FROM pg_locks l
WHERE l.pid = <worker_pid>;

-- Cancel with grace period
SELECT pg_background_cancel_v2_grace(<pid>, <cookie>, 5000);

-- Force cancel if grace period expires
SELECT pg_background_cancel_v2(<pid>, <cookie>);

-- ✅ Correct: Use CTE to consume once
WITH worker_results AS (
  SELECT * FROM pg_background_result_v2(<pid>, <cookie>) AS (col text)
)
SELECT * FROM worker_results
UNION ALL
SELECT * FROM worker_results;

-- Check DSM usage
SELECT count(*), sum(size) AS total_bytes
FROM pg_shmem_allocations
WHERE name LIKE '%dsm%';

-- Increase shared memory (postgresql.conf)
-- dynamic_shared_memory_type = posix  (or sysv, mmap)
-- Restart PostgreSQL

-- Always set statement_timeout on Windows
ALTER DATABASE mydb SET statement_timeout = '5min';

-- Or per-worker:
SELECT pg_background_launch_v2($$
  SET statement_timeout = '5min';
  SELECT slow_function();
$$);

-- Enable verbose logging
SET client_min_messages = DEBUG1;
SET log_min_messages = DEBUG1;

-- Launch worker (check logs for DSM info)
SELECT * FROM pg_background_launch_v2('SELECT 1') AS h \gset;

-- Check PostgreSQL logs for:
-- - "registered dynamic background worker"
-- - "DSM segment attached"
-- - Worker execution details

SELECT pg_background_launch_v2($$
  SELECT * FROM dblink('dbname=other_db', 'SELECT ...')
$$);

-- Store results in temp table
CREATE TEMP TABLE worker_output AS
  SELECT * FROM pg_background_result_v2(<pid>, <cookie>) AS (col text);

-- Query multiple times
SELECT * FROM worker_output WHERE col LIKE '%foo%';
SELECT count(*) FROM worker_output;

SELECT
  pid,
  backend_type,
  state,
  query,
  backend_start
FROM pg_stat_activity
WHERE backend_type LIKE '%background%';

SELECT * FROM pg_background_launch_v2('...') AS h \gset
SELECT pg_background_result_v2(:'h.pid', :'h.cookie');

SELECT pg_background_launch('...') AS pid \gset  -- No PID reuse protection
SELECT pg_background_result(:pid);

-- Database-wide default
ALTER DATABASE production SET statement_timeout = '10min';

-- Or per-worker
SELECT pg_background_launch_v2($$
  SET statement_timeout = '5min';
  SELECT slow_query();
$$);

-- ✅ Idiomatic: submit + detach
SELECT * FROM pg_background_submit_v2('INSERT INTO log ...') AS h \gset;
SELECT pg_background_detach_v2(:'h.pid', :'h.cookie');

-- ❌ Verbose: launch + detach without result retrieval
SELECT * FROM pg_background_launch_v2('INSERT INTO log ...') AS h \gset;
SELECT pg_background_detach_v2(:'h.pid', :'h.cookie');

-- Scheduled cleanup of stale workers
CREATE OR REPLACE FUNCTION cleanup_stale_workers()
RETURNS void AS $$
DECLARE r record;
BEGIN
  FOR r IN
    SELECT *
    FROM pg_background_list_v2() AS (
      pid int4, cookie int8, launched_at timestamptz, user_id oid,
      queue_size int4, state text, sql_preview text, last_error text, consumed bool
    )
    WHERE state IN ('stopped', 'error')
      AND (now() - launched_at) > interval '1 hour'
  LOOP
    PERFORM pg_background_detach_v2(r.pid, r.cookie);
  END LOOP;
END;
$$ LANGUAGE plpgsql;

-- Run periodically
SELECT cleanup_stale_workers();

-- ✅ Safe: Use format() with %L
CREATE FUNCTION safe_worker(table_name text) RETURNS void AS $$
BEGIN
  PERFORM pg_background_launch_v2(
    format('VACUUM %I', table_name)  -- %I for identifiers
  );
END;
$$ LANGUAGE plpgsql;

DO $$
DECLARE
  h pg_background_handle;
  result_val text;
BEGIN
  SELECT * INTO h FROM pg_background_launch_v2('SELECT 1/0');
  
  BEGIN
    SELECT * INTO result_val FROM pg_background_result_v2(h.pid, h.cookie) AS (r text);
  EXCEPTION WHEN OTHERS THEN
    RAISE NOTICE 'Worker failed: %', SQLERRM;
    -- Cleanup
    PERFORM pg_background_detach_v2(h.pid, h.cookie);
  END;
END;
$$;

-- ✅ Good: Clear intent
SELECT * FROM pg_background_launch_v2($$
  /* Background VACUUM for nightly maintenance */
  VACUUM (VERBOSE, ANALYZE) user_activity;
$$) AS h \gset;

-- Comment visible in list_v2() sql_preview

-- Simulate crash: check handle invalidation
SELECT * FROM pg_background_launch_v2('SELECT pg_sleep(100)') AS h \gset;
-- Restart PostgreSQL
SELECT pg_background_wait_v2(:'h.pid', :'h.cookie');  -- Should error gracefully

ALTER EXTENSION pg_background UPDATE TO '1.6';

-- Multi-hop upgrade path
ALTER EXTENSION pg_background UPDATE TO '1.4';
ALTER EXTENSION pg_background UPDATE TO '1.6';

SELECT pg_background_launch('VACUUM my_table') AS pid \gset
SELECT pg_background_detach(:pid);

SELECT * FROM pg_background_submit_v2('VACUUM my_table') AS h \gset;
SELECT pg_background_detach_v2(:'h.pid', :'h.cookie');