-- 15_production_hardening.sql: DLQ cleanup, backpressure function, autovacuum (TRD-06)
-- Note: Backpressure edge cases (at limit, above limit) tested via e2e with ALTER SYSTEM.
-- GUC max_queue_size is PGC_SIGHUP so cannot be SET in regression tests.
-- ============================================================================
-- SETUP
-- ============================================================================
SELECT ulak.create_endpoint('ph_ep', 'http',
    '{"url": "http://localhost:9999/ph"}'::jsonb) IS NOT NULL AS ep_created;
 ep_created 
------------
 t
(1 row)

CREATE TEMP TABLE ph_ids AS
SELECT id AS endpoint_id FROM ulak.endpoints WHERE name = 'ph_ep';
-- ============================================================================
-- CLEANUP_DLQ: EMPTY TABLE RETURNS 0
-- ============================================================================
SELECT ulak.cleanup_dlq() AS dlq_cleanup_empty;
 dlq_cleanup_empty 
-------------------
                 0
(1 row)

-- ============================================================================
-- CLEANUP_DLQ: RETENTION BOUNDARY (default 30 days)
-- ============================================================================
-- 3 recent entries (within retention)
INSERT INTO ulak.dlq (
    original_message_id, endpoint_id, endpoint_name, protocol,
    payload, retry_count, original_created_at, failed_at, archived_at
)
SELECT
    1000 + s, (SELECT endpoint_id FROM ph_ids), 'ph_ep', 'http',
    format('{"dlq_recent": %s}', s)::jsonb, 3,
    NOW() - interval '1 day', NOW() - interval '1 day',
    NOW() - (s || ' days')::interval
FROM generate_series(1, 3) AS s;
-- 2 old entries (beyond 30d retention)
INSERT INTO ulak.dlq (
    original_message_id, endpoint_id, endpoint_name, protocol,
    payload, retry_count, original_created_at, failed_at, archived_at
)
SELECT
    2000 + s, (SELECT endpoint_id FROM ph_ids), 'ph_ep', 'http',
    format('{"dlq_old": %s}', s)::jsonb, 5,
    NOW() - interval '90 days', NOW() - interval '90 days',
    NOW() - ((30 + s) || ' days')::interval
FROM generate_series(1, 2) AS s;
SELECT count(*) AS dlq_before FROM ulak.dlq
WHERE endpoint_id = (SELECT endpoint_id FROM ph_ids);
 dlq_before 
------------
          5
(1 row)

SELECT ulak.cleanup_dlq() AS dlq_cleaned;
 dlq_cleaned 
-------------
           2
(1 row)

SELECT count(*) AS dlq_after FROM ulak.dlq
WHERE endpoint_id = (SELECT endpoint_id FROM ph_ids);
 dlq_after 
-----------
         3
(1 row)

DELETE FROM ulak.dlq WHERE endpoint_id = (SELECT endpoint_id FROM ph_ids);
-- ============================================================================
-- _CHECK_BACKPRESSURE: FUNCTION EXISTS AND WORKS (default limit 1M, queue empty)
-- ============================================================================
-- With default max_queue_size=1000000 and empty queue, should pass
SELECT ulak._check_backpressure();
 _check_backpressure 
---------------------
 
(1 row)

-- Large projected insert count should be rejected even when queue is currently empty
DO $$
BEGIN
    PERFORM ulak._check_backpressure(1000001);
EXCEPTION WHEN OTHERS THEN
    RAISE NOTICE 'projected backpressure: rejected';
END $$;
NOTICE:  projected backpressure: rejected
-- Verify function exists with correct signature
SELECT proname, pronargs FROM pg_proc
WHERE proname = '_check_backpressure'
  AND pronamespace = (SELECT oid FROM pg_namespace WHERE nspname = 'ulak');
       proname       | pronargs 
---------------------+----------
 _check_backpressure |        1
(1 row)

-- ============================================================================
-- DLQ AUTOVACUUM SETTINGS
-- ============================================================================
SELECT
    (reloptions @> ARRAY['autovacuum_vacuum_scale_factor=0.05']) AS dlq_vacuum_sf,
    (reloptions @> ARRAY['autovacuum_analyze_scale_factor=0.05']) AS dlq_analyze_sf
FROM pg_class WHERE oid = 'ulak.dlq'::regclass;
 dlq_vacuum_sf | dlq_analyze_sf 
---------------+----------------
 t             | t
(1 row)

-- ============================================================================
-- QUEUE AUTOVACUUM SETTINGS
-- ============================================================================
SELECT
    (reloptions @> ARRAY['autovacuum_vacuum_scale_factor=0.01']) AS queue_vacuum_sf,
    (reloptions @> ARRAY['autovacuum_vacuum_cost_delay=2']) AS queue_cost_delay
FROM pg_class WHERE oid = 'ulak.queue'::regclass;
 queue_vacuum_sf | queue_cost_delay 
-----------------+------------------
 t               | t
(1 row)

-- ============================================================================
-- GUC DEFAULTS VERIFICATION
-- ============================================================================
SELECT current_setting('ulak.dlq_retention_days') AS dlq_retention;
 dlq_retention 
---------------
 30
(1 row)

SELECT current_setting('ulak.archive_retention_months') AS archive_retention;
 archive_retention 
-------------------
 6
(1 row)

SELECT current_setting('ulak.max_queue_size') AS max_queue;
 max_queue 
-----------
 1000000
(1 row)

-- ============================================================================
-- MONITORING SURFACE
-- ============================================================================
SELECT count(*) = current_setting('ulak.workers')::int AS worker_row_count
FROM ulak.get_worker_status();
 worker_row_count 
------------------
 t
(1 row)

SELECT bool_and(state IN ('running', 'stopped')) AS worker_states_valid
FROM ulak.get_worker_status();
 worker_states_valid 
---------------------
 t
(1 row)

SELECT count(*) = 2 AS health_component_count
FROM ulak.health_check();
 health_component_count 
------------------------
 t
(1 row)

-- ============================================================================
-- ENQUEUE LIMITS COME FROM THE SERVER SETTINGS (ulak._limit_setting)
-- ============================================================================
SELECT ulak._limit_setting('max_payload_size') AS max_payload_size,
       ulak._limit_setting('max_queue_size') AS max_queue_size;
 max_payload_size | max_queue_size 
------------------+----------------
          1048576 |        1000000
(1 row)

-- Both are SIGHUP settings: a session cannot raise them for itself
DO $$
BEGIN
    PERFORM set_config('ulak.max_payload_size', '104857600', false);
    RAISE NOTICE 'session SET accepted (unexpected)';
EXCEPTION WHEN cant_change_runtime_param THEN
    RAISE NOTICE 'session SET of max_payload_size refused (%)', SQLSTATE;
END $$;
NOTICE:  session SET of max_payload_size refused (55P02)
DO $$
BEGIN
    PERFORM ulak._limit_setting('max_retries');
EXCEPTION WHEN invalid_parameter_value THEN
    RAISE NOTICE 'unknown limit refused: %', SQLERRM;
END $$;
NOTICE:  unknown limit refused: unknown ulak limit "max_retries"
-- ============================================================================
-- PAYLOAD LIMIT ON EVERY ENTRY POINT (default ulak.max_payload_size = 1 MB)
-- ============================================================================
SELECT ulak.create_event_type('ph_evt') IS NOT NULL AS evt_created;
 evt_created 
-------------
 t
(1 row)

SELECT ulak.subscribe('ph_evt', 'ph_ep') IS NOT NULL AS sub_created;
 sub_created 
-------------
 t
(1 row)

CREATE TEMP TABLE ph_big AS
SELECT jsonb_build_object('d', repeat('x', 1100000)) AS p;
DO $$
DECLARE
    v_big jsonb := (SELECT p FROM ph_big);
    v_calls text[] := ARRAY[
        'send', 'send_with_options', 'send_batch', 'send_batch_with_priority',
        'publish', 'publish_batch'];
    v_call text;
BEGIN
    FOREACH v_call IN ARRAY v_calls LOOP
        BEGIN
            CASE v_call
                WHEN 'send' THEN PERFORM ulak.send('ph_ep', v_big);
                WHEN 'send_with_options' THEN PERFORM ulak.send_with_options('ph_ep', v_big);
                WHEN 'send_batch' THEN
                    PERFORM ulak.send_batch('ph_ep', ARRAY['{"small": 1}'::jsonb, v_big]);
                WHEN 'send_batch_with_priority' THEN
                    PERFORM ulak.send_batch_with_priority('ph_ep', ARRAY[v_big], 5::smallint);
                WHEN 'publish' THEN PERFORM ulak.publish('ph_evt', v_big);
                WHEN 'publish_batch' THEN
                    PERFORM ulak.publish_batch(jsonb_build_array(
                        jsonb_build_object('event_type', 'ph_evt', 'payload', '{"small": 1}'::jsonb),
                        jsonb_build_object('event_type', 'ph_evt', 'payload', v_big)));
            END CASE;
            RAISE NOTICE '%: accepted (unexpected)', v_call;
        EXCEPTION WHEN program_limit_exceeded THEN
            RAISE NOTICE '%: rejected (%)', v_call, SQLSTATE;
        END;
    END LOOP;
END $$;
NOTICE:  send: rejected (54000)
NOTICE:  send_with_options: rejected (54000)
NOTICE:  send_batch: rejected (54000)
NOTICE:  send_batch_with_priority: rejected (54000)
NOTICE:  publish: rejected (54000)
NOTICE:  publish_batch: rejected (54000)
-- Nothing from the rejected calls reached the queue
SELECT count(*) AS oversized_rows FROM ulak.queue
WHERE endpoint_id = (SELECT endpoint_id FROM ph_ids)
  AND octet_length(payload::text) > 1000000;
 oversized_rows 
----------------
              0
(1 row)

-- Payloads under the limit are still accepted
SELECT cardinality(ulak.send_batch('ph_ep', ARRAY['{"a": 1}'::jsonb, '{"b": 2}'::jsonb])) AS small_batch;
 small_batch 
-------------
           2
(1 row)

SELECT ulak.publish('ph_evt', '{"small": true}'::jsonb) AS published;
 published 
-----------
         1
(1 row)

DROP TABLE ph_big;
-- ============================================================================
-- BACKPRESSURE COUNTS THE WHOLE BATCH (default ulak.max_queue_size = 1000000)
-- ============================================================================
-- A batch larger than the limit is refused even though the queue is nearly empty
DO $$
DECLARE
    v_payloads jsonb[] := array_fill('{}'::jsonb, ARRAY[1000001]);
BEGIN
    BEGIN
        PERFORM ulak.send_batch('ph_ep', v_payloads);
        RAISE NOTICE 'send_batch: accepted (unexpected)';
    EXCEPTION WHEN configuration_limit_exceeded THEN
        RAISE NOTICE 'send_batch: rejected (%)', SQLSTATE;
    END;
    BEGIN
        PERFORM ulak.send_batch_with_priority('ph_ep', v_payloads, 1::smallint);
        RAISE NOTICE 'send_batch_with_priority: accepted (unexpected)';
    EXCEPTION WHEN configuration_limit_exceeded THEN
        RAISE NOTICE 'send_batch_with_priority: rejected (%)', SQLSTATE;
    END;
END $$;
NOTICE:  send_batch: rejected (53400)
NOTICE:  send_batch_with_priority: rejected (53400)
SELECT count(*) AS queued_after_backpressure FROM ulak.queue
WHERE endpoint_id = (SELECT endpoint_id FROM ph_ids)
  AND payload = '{}'::jsonb;
 queued_after_backpressure 
---------------------------
                         0
(1 row)

SELECT ulak.drop_event_type('ph_evt') AS evt_dropped;
 evt_dropped 
-------------
 t
(1 row)

-- ============================================================================
-- CLEANUP
-- ============================================================================
DELETE FROM ulak.queue WHERE endpoint_id = (SELECT endpoint_id FROM ph_ids);
DELETE FROM ulak.dlq WHERE endpoint_id = (SELECT endpoint_id FROM ph_ids);
DROP TABLE ph_ids;
SELECT ulak.drop_endpoint('ph_ep');
 drop_endpoint 
---------------
 t
(1 row)