-- Test: advanced SQL query types — recursive CTEs, LATERAL joins, window functions
\set VERBOSITY terse
\set ON_ERROR_STOP on
CREATE EXTENSION pg_reactive;
----------------------------------------------------------------------
-- Test 1: WITH RECURSIVE (tree traversal)
----------------------------------------------------------------------
CREATE TABLE tree_nodes (
    id serial PRIMARY KEY,
    parent_id int REFERENCES tree_nodes(id),
    name text
);
INSERT INTO tree_nodes (id, parent_id, name) VALUES
    (1, NULL, 'root'),
    (2, 1, 'child_a'),
    (3, 1, 'child_b'),
    (4, 2, 'grandchild_a1'),
    (5, 3, 'grandchild_b1');
SELECT setval('tree_nodes_id_seq', 5);
 setval 
--------
      5
(1 row)

-- Subscribe to recursive CTE
SELECT pgr.subscribe('adv_recursive', $q$
    WITH RECURSIVE tree AS (
        SELECT id, parent_id, name, name::text AS path
        FROM tree_nodes WHERE parent_id IS NULL
        UNION ALL
        SELECT tn.id, tn.parent_id, tn.name, t.path || ' > ' || tn.name
        FROM tree_nodes tn JOIN tree t ON tn.parent_id = t.id
    )
    SELECT id, name, path FROM tree ORDER BY id
$q$);
                                      subscribe                                      
-------------------------------------------------------------------------------------
 {"mode": "delta", "status": "subscribed", "tables": 1, "query_id": "adv_recursive"}
(1 row)

-- Verify snapshot has all 5 nodes
SELECT count(*) AS recursive_snap FROM pgr."_snap_adv_recursive";
 recursive_snap 
----------------
              5
(1 row)

-- Verify snapshot content is correct
SELECT * FROM pgr."_snap_adv_recursive" ORDER BY id;
 id |     name      |              path              
----+---------------+--------------------------------
  1 | root          | root
  2 | child_a       | root > child_a
  3 | child_b       | root > child_b
  4 | grandchild_a1 | root > child_a > grandchild_a1
  5 | grandchild_b1 | root > child_b > grandchild_b1
(5 rows)

-- INSERT a new grandchild — should add one row to the tree
INSERT INTO tree_nodes (id, parent_id, name) VALUES (6, 2, 'grandchild_a2');
-- Snapshot should now have 6 rows
SELECT count(*) AS recursive_after_insert FROM pgr."_snap_adv_recursive";
 recursive_after_insert 
------------------------
                      6
(1 row)

-- Verify the new row has correct path
SELECT name, path FROM pgr."_snap_adv_recursive" WHERE id = 6;
     name      |              path              
---------------+--------------------------------
 grandchild_a2 | root > child_a > grandchild_a2
(1 row)

-- Verify snapshot matches query exactly (zero diff)
SELECT count(*) AS recursive_diff FROM (
    (WITH RECURSIVE tree AS (
        SELECT id, parent_id, name, name::text AS path
        FROM tree_nodes WHERE parent_id IS NULL
        UNION ALL
        SELECT tn.id, tn.parent_id, tn.name, t.path || ' > ' || tn.name
        FROM tree_nodes tn JOIN tree t ON tn.parent_id = t.id
    )
    SELECT id, name, path FROM tree ORDER BY id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_recursive")
) t;
 recursive_diff 
----------------
              0
(1 row)

-- Test 1 bidirectional diff: verify snapshot has no stale rows
SELECT count(*) AS recursive_bidi_diff FROM (
    (SELECT id, name, path FROM pgr."_snap_adv_recursive")
    EXCEPT
    (WITH RECURSIVE tree AS (
        SELECT id, parent_id, name, name::text AS path
        FROM tree_nodes WHERE parent_id IS NULL
        UNION ALL
        SELECT tn.id, tn.parent_id, tn.name, t.path || ' > ' || tn.name
        FROM tree_nodes tn JOIN tree t ON tn.parent_id = t.id
    )
    SELECT id, name, path FROM tree ORDER BY id)
) t;
 recursive_bidi_diff 
---------------------
                   0
(1 row)

-- DELETE an intermediate node's child — should remove it from path results
DELETE FROM tree_nodes WHERE id = 5;
SELECT count(*) AS recursive_after_delete FROM pgr."_snap_adv_recursive";
 recursive_after_delete 
------------------------
                      5
(1 row)

-- Verify snapshot still matches query
SELECT count(*) AS recursive_diff2 FROM (
    (WITH RECURSIVE tree AS (
        SELECT id, parent_id, name, name::text AS path
        FROM tree_nodes WHERE parent_id IS NULL
        UNION ALL
        SELECT tn.id, tn.parent_id, tn.name, t.path || ' > ' || tn.name
        FROM tree_nodes tn JOIN tree t ON tn.parent_id = t.id
    )
    SELECT id, name, path FROM tree ORDER BY id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_recursive")
) t;
 recursive_diff2 
-----------------
               0
(1 row)

-- Cleanup
SELECT pgr.unsubscribe('adv_recursive');
 unsubscribe 
-------------
 t
(1 row)

DROP TABLE tree_nodes;
----------------------------------------------------------------------
-- Test 2: LATERAL join
----------------------------------------------------------------------
CREATE TABLE players (id serial PRIMARY KEY, name text);
CREATE TABLE scores (id serial PRIMARY KEY, player_id int REFERENCES players(id), score int, created_at timestamp DEFAULT now());
INSERT INTO players (id, name) VALUES (1, 'alice'), (2, 'bob'), (3, 'carol');
SELECT setval('players_id_seq', 3);
 setval 
--------
      3
(1 row)

INSERT INTO scores (player_id, score, created_at) VALUES
    (1, 100, '2025-01-01'),
    (1, 150, '2025-01-02'),
    (2, 200, '2025-01-01'),
    (2, 180, '2025-01-02'),
    (3, 90,  '2025-01-01');
-- Subscribe to LATERAL subquery — latest score per player
SELECT pgr.subscribe('adv_lateral', $q$
    SELECT p.id, p.name, latest.score
    FROM players p,
    LATERAL (SELECT score FROM scores WHERE player_id = p.id ORDER BY created_at DESC LIMIT 1) latest
    ORDER BY p.id
$q$);
                                     subscribe                                     
-----------------------------------------------------------------------------------
 {"mode": "delta", "status": "subscribed", "tables": 2, "query_id": "adv_lateral"}
(1 row)

-- Verify snapshot: 3 players, each with their latest score
SELECT count(*) AS lateral_snap FROM pgr."_snap_adv_lateral";
 lateral_snap 
--------------
            3
(1 row)

SELECT * FROM pgr."_snap_adv_lateral" ORDER BY id;
 id | name  | score 
----+-------+-------
  1 | alice |   150
  2 | bob   |   180
  3 | carol |    90
(3 rows)

-- Insert a new high score for alice — should update her latest
INSERT INTO scores (player_id, score, created_at) VALUES (1, 300, '2025-01-03');
-- Verify snapshot updated
SELECT * FROM pgr."_snap_adv_lateral" ORDER BY id;
 id | name  | score 
----+-------+-------
  1 | alice |   300
  2 | bob   |   180
  3 | carol |    90
(3 rows)

-- Verify zero diff
SELECT count(*) AS lateral_diff FROM (
    (SELECT p.id, p.name, latest.score
     FROM players p,
     LATERAL (SELECT score FROM scores WHERE player_id = p.id ORDER BY created_at DESC LIMIT 1) latest
     ORDER BY p.id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_lateral")
) t;
 lateral_diff 
--------------
            0
(1 row)

-- Add a new player with a score — LATERAL should include them
INSERT INTO players (id, name) VALUES (4, 'dave');
INSERT INTO scores (player_id, score, created_at) VALUES (4, 250, '2025-01-01');
SELECT count(*) AS lateral_after_new_player FROM pgr."_snap_adv_lateral";
 lateral_after_new_player 
--------------------------
                        4
(1 row)

-- Verify zero diff
SELECT count(*) AS lateral_diff2 FROM (
    (SELECT p.id, p.name, latest.score
     FROM players p,
     LATERAL (SELECT score FROM scores WHERE player_id = p.id ORDER BY created_at DESC LIMIT 1) latest
     ORDER BY p.id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_lateral")
) t;
 lateral_diff2 
---------------
             0
(1 row)

-- Cleanup
SELECT pgr.unsubscribe('adv_lateral');
 unsubscribe 
-------------
 t
(1 row)

DROP TABLE scores;
DROP TABLE players;
----------------------------------------------------------------------
-- Test 3: Window functions (ROW_NUMBER, DENSE_RANK)
----------------------------------------------------------------------
CREATE TABLE products (id serial PRIMARY KEY, title text, author_id int, price numeric);
INSERT INTO products (id, title, author_id, price) VALUES
    (1, 'Widget A', 10, 5.00),
    (2, 'Widget B', 10, 8.00),
    (3, 'Gadget C', 20, 3.00),
    (4, 'Gadget D', 20, 12.00);
SELECT setval('products_id_seq', 4);
 setval 
--------
      4
(1 row)

-- Subscribe to window function query
SELECT pgr.subscribe('adv_window', $q$
    SELECT id, title, author_id,
           row_number() OVER (ORDER BY id) AS rn,
           dense_rank() OVER (PARTITION BY author_id ORDER BY id) AS dr
    FROM products ORDER BY id
$q$);
                                    subscribe                                     
----------------------------------------------------------------------------------
 {"mode": "delta", "status": "subscribed", "tables": 1, "query_id": "adv_window"}
(1 row)

SELECT count(*) AS window_snap FROM pgr."_snap_adv_window";
 window_snap 
-------------
           4
(1 row)

SELECT * FROM pgr."_snap_adv_window" ORDER BY id;
 id |  title   | author_id | rn | dr 
----+----------+-----------+----+----
  1 | Widget A |        10 |  1 |  1
  2 | Widget B |        10 |  2 |  2
  3 | Gadget C |        20 |  3 |  1
  4 | Gadget D |        20 |  4 |  2
(4 rows)

-- INSERT new product — shifts row_number for all subsequent rows
INSERT INTO products (id, title, author_id, price) VALUES (5, 'Gadget E', 20, 7.00);
-- Verify snapshot matches query (window functions recalculated)
SELECT count(*) AS window_diff FROM (
    (SELECT id, title, author_id,
            row_number() OVER (ORDER BY id) AS rn,
            dense_rank() OVER (PARTITION BY author_id ORDER BY id) AS dr
     FROM products ORDER BY id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_window")
) t;
 window_diff 
-------------
           0
(1 row)

SELECT count(*) AS window_after_insert FROM pgr."_snap_adv_window";
 window_after_insert 
---------------------
                   5
(1 row)

-- DELETE a product — window values shift again
DELETE FROM products WHERE id = 2;
SELECT count(*) AS window_diff2 FROM (
    (SELECT id, title, author_id,
            row_number() OVER (ORDER BY id) AS rn,
            dense_rank() OVER (PARTITION BY author_id ORDER BY id) AS dr
     FROM products ORDER BY id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_window")
) t;
 window_diff2 
--------------
            0
(1 row)

SELECT count(*) AS window_after_delete FROM pgr."_snap_adv_window";
 window_after_delete 
---------------------
                   4
(1 row)

-- Cleanup
SELECT pgr.unsubscribe('adv_window');
 unsubscribe 
-------------
 t
(1 row)

DROP TABLE products;
----------------------------------------------------------------------
-- Test 4: Window functions with LAG/LEAD
----------------------------------------------------------------------
CREATE TABLE events (id serial PRIMARY KEY, title text, event_date date);
INSERT INTO events (id, title, event_date) VALUES
    (1, 'Alpha', '2025-01-01'),
    (2, 'Beta',  '2025-02-01'),
    (3, 'Gamma', '2025-03-01');
SELECT setval('events_id_seq', 3);
 setval 
--------
      3
(1 row)

-- Subscribe to LAG/LEAD query
SELECT pgr.subscribe('adv_laglead', $q$
    SELECT id, title,
           lag(title) OVER (ORDER BY id) AS prev_title,
           lead(title) OVER (ORDER BY id) AS next_title
    FROM events ORDER BY id
$q$);
                                     subscribe                                     
-----------------------------------------------------------------------------------
 {"mode": "delta", "status": "subscribed", "tables": 1, "query_id": "adv_laglead"}
(1 row)

SELECT count(*) AS laglead_snap FROM pgr."_snap_adv_laglead";
 laglead_snap 
--------------
            3
(1 row)

SELECT * FROM pgr."_snap_adv_laglead" ORDER BY id;
 id | title | prev_title | next_title 
----+-------+------------+------------
  1 | Alpha |            | Beta
  2 | Beta  | Alpha      | Gamma
  3 | Gamma | Beta       | 
(3 rows)

-- INSERT a new event — shifts lead/lag for neighbors
INSERT INTO events (id, title, event_date) VALUES (4, 'Delta', '2025-04-01');
-- Verify snapshot matches (Gamma now has next_title = 'Delta')
SELECT count(*) AS laglead_diff FROM (
    (SELECT id, title,
            lag(title) OVER (ORDER BY id) AS prev_title,
            lead(title) OVER (ORDER BY id) AS next_title
     FROM events ORDER BY id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_laglead")
) t;
 laglead_diff 
--------------
            0
(1 row)

SELECT count(*) AS laglead_after_insert FROM pgr."_snap_adv_laglead";
 laglead_after_insert 
----------------------
                    4
(1 row)

-- Verify specific values after insert
SELECT title, prev_title, next_title FROM pgr."_snap_adv_laglead" WHERE id = 3;
 title | prev_title | next_title 
-------+------------+------------
 Gamma | Beta       | Delta
(1 row)

-- DELETE middle event — shifts lag/lead for remaining
DELETE FROM events WHERE id = 2;
SELECT count(*) AS laglead_diff2 FROM (
    (SELECT id, title,
            lag(title) OVER (ORDER BY id) AS prev_title,
            lead(title) OVER (ORDER BY id) AS next_title
     FROM events ORDER BY id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_laglead")
) t;
 laglead_diff2 
---------------
             0
(1 row)

SELECT count(*) AS laglead_after_delete FROM pgr."_snap_adv_laglead";
 laglead_after_delete 
----------------------
                    3
(1 row)

-- After deleting Beta: Alpha(prev=NULL, next=Gamma), Gamma(prev=Alpha, next=Delta), Delta(prev=Gamma, next=NULL)
SELECT title, prev_title, next_title FROM pgr."_snap_adv_laglead" ORDER BY id;
 title | prev_title | next_title 
-------+------------+------------
 Alpha |            | Gamma
 Gamma | Alpha      | Delta
 Delta | Gamma      | 
(3 rows)

-- Cleanup
SELECT pgr.unsubscribe('adv_laglead');
 unsubscribe 
-------------
 t
(1 row)

DROP TABLE events;
----------------------------------------------------------------------
-- Test 5: Multiple window functions + aggregate window
----------------------------------------------------------------------
CREATE TABLE sales (id serial PRIMARY KEY, region text, amount numeric);
INSERT INTO sales (id, region, amount) VALUES
    (1, 'east', 100),
    (2, 'east', 200),
    (3, 'west', 150),
    (4, 'west', 300);
SELECT setval('sales_id_seq', 4);
 setval 
--------
      4
(1 row)

-- Subscribe to running total + partition rank
SELECT pgr.subscribe('adv_agg_window', $q$
    SELECT id, region, amount,
           sum(amount) OVER (PARTITION BY region ORDER BY id) AS running_total,
           rank() OVER (PARTITION BY region ORDER BY amount DESC) AS region_rank
    FROM sales ORDER BY id
$q$);
                                      subscribe                                       
--------------------------------------------------------------------------------------
 {"mode": "delta", "status": "subscribed", "tables": 1, "query_id": "adv_agg_window"}
(1 row)

SELECT count(*) AS agg_window_snap FROM pgr."_snap_adv_agg_window";
 agg_window_snap 
-----------------
               4
(1 row)

SELECT * FROM pgr."_snap_adv_agg_window" ORDER BY id;
 id | region | amount | running_total | region_rank 
----+--------+--------+---------------+-------------
  1 | east   |    100 |           100 |           2
  2 | east   |    200 |           300 |           1
  3 | west   |    150 |           150 |           2
  4 | west   |    300 |           450 |           1
(4 rows)

-- INSERT new sale — running totals and ranks recalculate
INSERT INTO sales (id, region, amount) VALUES (5, 'east', 500);
SELECT count(*) AS agg_window_diff FROM (
    (SELECT id, region, amount,
            sum(amount) OVER (PARTITION BY region ORDER BY id) AS running_total,
            rank() OVER (PARTITION BY region ORDER BY amount DESC) AS region_rank
     FROM sales ORDER BY id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_agg_window")
) t;
 agg_window_diff 
-----------------
               0
(1 row)

-- Verify the new row appears with correct running total
SELECT id, region, amount, running_total FROM pgr."_snap_adv_agg_window" WHERE id = 5;
 id | region | amount | running_total 
----+--------+--------+---------------
  5 | east   |    500 |           800
(1 row)

-- Cleanup
SELECT pgr.unsubscribe('adv_agg_window');
 unsubscribe 
-------------
 t
(1 row)

DROP TABLE sales;
----------------------------------------------------------------------
-- Test 6: CASE expression with aggregate
----------------------------------------------------------------------
CREATE TABLE orders (id serial PRIMARY KEY, status text, total numeric);
INSERT INTO orders (id, status, total) VALUES
    (1, 'complete', 100),
    (2, 'pending',  50),
    (3, 'complete', 200),
    (4, 'cancelled', 75);
SELECT setval('orders_id_seq', 4);
 setval 
--------
      4
(1 row)

SELECT pgr.subscribe('adv_case', $q$
    SELECT status,
           count(*) AS cnt,
           CASE WHEN count(*) > 1 THEN 'high' ELSE 'low' END AS volume
    FROM orders
    GROUP BY status
    ORDER BY status
$q$);
                                   subscribe                                    
--------------------------------------------------------------------------------
 {"mode": "delta", "status": "subscribed", "tables": 1, "query_id": "adv_case"}
(1 row)

SELECT count(*) AS case_snap FROM pgr."_snap_adv_case";
 case_snap 
-----------
         3
(1 row)

SELECT * FROM pgr."_snap_adv_case" ORDER BY status;
  status   | cnt | volume 
-----------+-----+--------
 cancelled |   1 | low
 complete  |   2 | high
 pending   |   1 | low
(3 rows)

-- Add another pending order — pending count goes from 1 to 2
INSERT INTO orders (id, status, total) VALUES (5, 'pending', 80);
-- Verify snapshot updated
SELECT count(*) AS case_diff FROM (
    (SELECT status,
            count(*) AS cnt,
            CASE WHEN count(*) > 1 THEN 'high' ELSE 'low' END AS volume
     FROM orders GROUP BY status ORDER BY status)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_case")
) t;
 case_diff 
-----------
         0
(1 row)

-- Pending should now show 'high' volume
SELECT status, cnt, volume FROM pgr."_snap_adv_case" WHERE status = 'pending';
 status  | cnt | volume 
---------+-----+--------
 pending |   2 | high
(1 row)

-- Cleanup
SELECT pgr.unsubscribe('adv_case');
 unsubscribe 
-------------
 t
(1 row)

DROP TABLE orders;
----------------------------------------------------------------------
-- Test 7: COALESCE / NULLIF with LEFT JOIN
----------------------------------------------------------------------
CREATE TABLE departments (id serial PRIMARY KEY, name text);
CREATE TABLE employees (id serial PRIMARY KEY, dept_id int REFERENCES departments(id), name text, salary numeric);
INSERT INTO departments (id, name) VALUES (1, 'Engineering'), (2, 'Marketing'), (3, 'Sales');
SELECT setval('departments_id_seq', 3);
 setval 
--------
      3
(1 row)

INSERT INTO employees (dept_id, name, salary) VALUES
    (1, 'Alice', 90000),
    (1, 'Bob',   80000),
    (2, 'Carol', 70000);
SELECT pgr.subscribe('adv_coalesce', $q$
    SELECT d.id, d.name AS dept,
           COALESCE(count(e.id), 0) AS headcount,
           COALESCE(avg(e.salary), 0) AS avg_salary
    FROM departments d
    LEFT JOIN employees e ON e.dept_id = d.id
    GROUP BY d.id, d.name
    ORDER BY d.id
$q$);
                                     subscribe                                      
------------------------------------------------------------------------------------
 {"mode": "delta", "status": "subscribed", "tables": 2, "query_id": "adv_coalesce"}
(1 row)

SELECT count(*) AS coalesce_snap FROM pgr."_snap_adv_coalesce";
 coalesce_snap 
---------------
             3
(1 row)

SELECT * FROM pgr."_snap_adv_coalesce" ORDER BY id;
 id |    dept     | headcount |     avg_salary     
----+-------------+-----------+--------------------
  1 | Engineering |         2 | 85000.000000000000
  2 | Marketing   |         1 | 70000.000000000000
  3 | Sales       |         0 |                  0
(3 rows)

-- Add an employee to the empty Sales dept
INSERT INTO employees (dept_id, name, salary) VALUES (3, 'Dave', 60000);
SELECT count(*) AS coalesce_diff FROM (
    (SELECT d.id, d.name AS dept,
            COALESCE(count(e.id), 0) AS headcount,
            COALESCE(avg(e.salary), 0) AS avg_salary
     FROM departments d
     LEFT JOIN employees e ON e.dept_id = d.id
     GROUP BY d.id, d.name ORDER BY d.id)
    EXCEPT
    (SELECT * FROM pgr."_snap_adv_coalesce")
) t;
 coalesce_diff 
---------------
             0
(1 row)

-- Sales dept should now show headcount=1
SELECT dept, headcount FROM pgr."_snap_adv_coalesce" WHERE id = 3;
 dept  | headcount 
-------+-----------
 Sales |         1
(1 row)

-- Cleanup
SELECT pgr.unsubscribe('adv_coalesce');
 unsubscribe 
-------------
 t
(1 row)

DROP TABLE employees;
DROP TABLE departments;
----------------------------------------------------------------------
-- Final: verify all subscriptions cleaned up
----------------------------------------------------------------------
SELECT count(*) AS remaining_subs FROM pgr.get_subscriptions()
WHERE query_id LIKE 'adv_%';
 remaining_subs 
----------------
              0
(1 row)

DROP EXTENSION pg_reactive;