-- Copyright (c) Microsoft Corporation.
-- Licensed under the PostgreSQL License.

-- Merged from: 30_graph_reuse, 32_invalid_node_type
-- Tests: graph storage and reuse, identical DSL produces identical JSON,
--        invalid node_type rejection by df.start() and df.explain()
SET SESSION AUTHORIZATION df_e2e_user;

-- === Test: 30_graph_reuse ===

DROP TABLE IF EXISTS test_graph_reuse;
CREATE TABLE test_graph_reuse (
    id SERIAL PRIMARY KEY,
    value INT
);

-- Test 1: Store a graph and start it later
CREATE TEMP TABLE _stored_graph AS
SELECT 'SELECT 1' ~> 'INSERT INTO test_graph_reuse (value) VALUES (1)' AS graph_json;

CREATE TEMP TABLE _test1_instance1 AS
SELECT df.start((SELECT graph_json FROM _stored_graph), 'test-reuse-1') AS instance_id;

DO $$
DECLARE
    inst_id TEXT;
    status TEXT;
    attempts INT := 0;
BEGIN
    SELECT instance_id INTO inst_id FROM _test1_instance1;

    LOOP
        SELECT s INTO status FROM df.status(inst_id) s;
        EXIT WHEN lower(status) IN ('completed', 'failed', 'cancelled') OR attempts > 300;
        PERFORM pg_sleep(0.1);
        attempts := attempts + 1;
    END LOOP;

    IF lower(status) != 'completed' THEN
        RAISE EXCEPTION 'TEST FAILED: First execution status = %', status;
    END IF;

    RAISE NOTICE 'First execution completed';
END $$;

CREATE TEMP TABLE _test1_instance2 AS
SELECT df.start((SELECT graph_json FROM _stored_graph), 'test-reuse-2') AS instance_id;

DO $$
DECLARE
    inst_id TEXT;
    status TEXT;
    attempts INT := 0;
BEGIN
    SELECT instance_id INTO inst_id FROM _test1_instance2;

    LOOP
        SELECT s INTO status FROM df.status(inst_id) s;
        EXIT WHEN lower(status) IN ('completed', 'failed', 'cancelled') OR attempts > 300;
        PERFORM pg_sleep(0.1);
        attempts := attempts + 1;
    END LOOP;

    IF lower(status) != 'completed' THEN
        RAISE EXCEPTION 'TEST FAILED: Second execution status = %', status;
    END IF;

    RAISE NOTICE 'Second execution completed';
END $$;

DO $$
BEGIN
    IF (SELECT COUNT(*) FROM test_graph_reuse) != 2 THEN
        RAISE EXCEPTION 'TEST FAILED: Expected 2 rows, got %', (SELECT COUNT(*) FROM test_graph_reuse);
    END IF;

    RAISE NOTICE 'Test 1 PASSED: Graph stored and reused successfully';
END $$;

DROP TABLE _stored_graph;
DROP TABLE _test1_instance1;
DROP TABLE _test1_instance2;

-- Test 2: Verify identical DSL expressions produce identical JSON
DO $$
DECLARE
    graph1 TEXT;
    graph2 TEXT;
BEGIN
    SELECT 'SELECT 2' ~> 'SELECT 3' INTO graph1;
    SELECT 'SELECT 2' ~> 'SELECT 3' INTO graph2;

    IF graph1 IS DISTINCT FROM graph2 THEN
        RAISE EXCEPTION 'TEST FAILED: Identical DSL expressions produced different JSON. Graph 1: %, Graph 2: %', graph1, graph2;
    END IF;

    RAISE NOTICE 'Test 2 PASSED: Identical DSL expressions produce identical Durofut JSON';
END $$;

-- Test 3: Different graphs should have different JSON
DO $$
DECLARE
    graph1 TEXT;
    graph2 TEXT;
BEGIN
    SELECT 'SELECT 4' ~> 'SELECT 5' INTO graph1;
    SELECT 'SELECT 6' ~> 'SELECT 7' INTO graph2;

    IF graph1 = graph2 THEN
        RAISE EXCEPTION 'TEST FAILED: Different graphs produced identical JSON: %', graph1;
    END IF;

    RAISE NOTICE 'Test 3 PASSED: Different graphs produce different Durofut JSON';
END $$;

DROP TABLE test_graph_reuse;

-- === Test: 32_invalid_node_type ===

DO $body$
DECLARE
    explanation TEXT;
BEGIN
    BEGIN
        PERFORM df.start('{"node_type":"NOT_A_NODE"}');
        RAISE EXCEPTION 'TEST FAILED: df.start should have rejected invalid node_type';
    EXCEPTION WHEN OTHERS THEN
        IF SQLERRM LIKE 'TEST FAILED:%' THEN
            RAISE;
        END IF;
        IF SQLERRM NOT LIKE '%Unknown node_type ''NOT_A_NODE''%' THEN
            RAISE EXCEPTION 'TEST FAILED: df.start returned the wrong error: %', SQLERRM;
        END IF;
        RAISE NOTICE 'Caught expected error: %', SQLERRM;
    END;

    explanation := df.explain('{"node_type":"NOT_A_NODE"}');
    IF explanation NOT LIKE 'Invalid durable function graph: root: Unknown node_type ''NOT_A_NODE''%' THEN
        RAISE EXCEPTION 'TEST FAILED: df.explain returned the wrong error: %', explanation;
    END IF;

    RAISE NOTICE 'TEST PASSED: invalid node_type handling';
END $body$;

-- === Test: await_instance blocked inside workflow ===
-- df.await_instance blocks the calling backend on a polling loop.
-- Inside a workflow that backend is a BGW thread, so calling it would
-- pin a worker slot for up to `timeout_seconds` and (if waiting on the
-- current instance) deadlock the workflow on itself. The function must
-- refuse to run when df.in_workflow='true'.

CREATE TEMP TABLE _test_wait_blocked (instance_id TEXT);

INSERT INTO _test_wait_blocked SELECT df.start(
    'SELECT df.await_instance(''nonexistent-instance'')',
    'test-wait-blocked'
);

DO $$
DECLARE
    inst_id TEXT;
    status TEXT;
    node_error TEXT;
BEGIN
    SELECT instance_id INTO inst_id FROM _test_wait_blocked;
    RAISE NOTICE 'Testing await_instance blocked in workflow: %', inst_id;

    SELECT df.await_instance(inst_id) INTO status;

    IF status != 'failed' THEN
        RAISE EXCEPTION 'TEST FAILED: expected workflow to fail but status = %', status;
    END IF;

    SELECT n.result::text INTO node_error
    FROM df.nodes n
    WHERE n.instance_id = inst_id AND n.status = 'failed'
    LIMIT 1;

    IF node_error NOT LIKE '%cannot be called inside a workflow%' THEN
        RAISE EXCEPTION 'TEST FAILED: expected "cannot be called inside a workflow" error, got: %', node_error;
    END IF;

    RAISE NOTICE 'TEST PASSED: await_instance_blocked_in_workflow';
END $$;

DROP TABLE _test_wait_blocked;

-- === Test: deep graph composition (#327) ===

DO $$
DECLARE
    graph TEXT := df.sql('SELECT 1');
    explanation TEXT;
BEGIN
    FOR i IN 1..129 LOOP
        graph := df.seq(graph, 'SELECT 1');
    END LOOP;

    explanation := df.explain(graph);
    IF explanation LIKE 'Invalid durable function graph:%'
         OR pg_catalog.regexp_count(explanation, '→') != 129 THEN
        RAISE EXCEPTION 'TEST FAILED: 129-level df.seq graph was corrupted: %', explanation;
    END IF;

    RAISE NOTICE 'TEST PASSED: df.seq composes beyond serde recursion limit';
END $$;

DO $$
DECLARE
    graph TEXT := df.sql('SELECT 1');
    explanation TEXT;
BEGIN
    FOR i IN 1..129 LOOP
        graph := ('SELECT true' ?> graph) !> 'SELECT 0';
    END LOOP;

    explanation := df.explain(graph);
    IF explanation LIKE 'Invalid durable function graph:%'
       OR pg_catalog.regexp_count(explanation, 'IF') != 129 THEN
        RAISE EXCEPTION 'TEST FAILED: 129-level ?>/!> graph was corrupted: %', explanation;
    END IF;

    RAISE NOTICE 'TEST PASSED: ?>/!> composes beyond parser recursion limit';
END $$;

DO $$
DECLARE
    condition_graph TEXT := df.sql('SELECT true');
    graph TEXT;
    explanation TEXT;
BEGIN
    -- 129 levels intentionally exceed serde_json's default 128-level recursion limit.
    FOR i IN 1..129 LOOP
        condition_graph := df.seq(condition_graph, 'SELECT true');
    END LOOP;

    graph := df.if(condition_graph, 'SELECT 1', 'SELECT 0');
    explanation := df.explain(graph);
    IF explanation LIKE 'Invalid durable function graph:%'
       OR pg_catalog.regexp_count(explanation, '→') != 129 THEN
        RAISE EXCEPTION 'TEST FAILED: deep condition graph was corrupted: %', explanation;
    END IF;

    RAISE NOTICE 'TEST PASSED: config children compose beyond serde recursion limit';
END $$;

DO $$
DECLARE
    graph TEXT := df.sql('SELECT 1');
    explanation TEXT;
BEGIN
    FOR i IN 1..256 LOOP
        graph := df.seq(graph, 'SELECT 1');
    END LOOP;

    explanation := df.explain(graph);
    IF explanation LIKE 'Invalid durable function graph:%' THEN
        RAISE EXCEPTION 'TEST FAILED: graph at maximum depth was rejected: %', explanation;
    END IF;

    graph := df.seq(graph, 'SELECT 1');
    explanation := df.explain(graph);
    IF explanation NOT LIKE '%maximum nesting depth of 256%' THEN
        RAISE EXCEPTION 'TEST FAILED: graph beyond maximum depth was not rejected cleanly: %', explanation;
    END IF;

    RAISE NOTICE 'TEST PASSED: df.explain enforces the configured depth boundary';
END $$;

CREATE TEMP TABLE _deep_graph_execution (instance_id TEXT);

DO $$
DECLARE
    graph TEXT := df.sql('SELECT 1');
BEGIN
    FOR i IN 2..200 LOOP
        graph := df.seq(graph, 'SELECT 1');
    END LOOP;

    INSERT INTO _deep_graph_execution SELECT df.start(graph, 'test-deep-graph-200');
END $$;

DO $$
DECLARE
    inst_id TEXT;
    status TEXT;
BEGIN
    SELECT instance_id INTO inst_id FROM _deep_graph_execution;
    SELECT df.await_instance(inst_id, 300) INTO status;

    IF status != 'completed' THEN
        RAISE EXCEPTION 'TEST FAILED: 200-step graph status = %', status;
    END IF;

    IF (SELECT count(*) FROM df.nodes WHERE instance_id = inst_id) != 399 THEN
        RAISE EXCEPTION 'TEST FAILED: 200-step graph did not materialize 399 nodes';
    END IF;

    RAISE NOTICE 'TEST PASSED: 200-step graph executes end to end';
END $$;

DROP TABLE _deep_graph_execution;

-- Exercise the node INSERT chunk boundary without committing a workflow that
-- would schedule 1,001 parallel SQL activities.
BEGIN;
CREATE TEMP TABLE _batched_graph_execution (instance_id TEXT);

INSERT INTO _batched_graph_execution
SELECT df.start(
    pg_catalog.jsonb_build_object(
        'node_type', 'JOIN',
        'left_node', pg_catalog.jsonb_build_object('node_type', 'SQL', 'query', 'SELECT 1'),
        'right_node', pg_catalog.jsonb_build_object('node_type', 'SQL', 'query', 'SELECT 1'),
        'extra_nodes', (
            SELECT pg_catalog.jsonb_agg(
                pg_catalog.jsonb_build_object('node_type', 'SQL', 'query', 'SELECT 1')
            )
            FROM pg_catalog.generate_series(1, 999)
        )
    )::TEXT,
    'test-batched-node-insert'
);

DO $$
DECLARE
    inst_id TEXT;
    node_total BIGINT;
BEGIN
    SELECT instance_id INTO inst_id FROM _batched_graph_execution;
    SELECT count(*) INTO node_total FROM df.nodes WHERE instance_id = inst_id;

    IF node_total != 1002 THEN
        RAISE EXCEPTION 'TEST FAILED: batched graph materialized % nodes instead of 1002', node_total;
    END IF;

    RAISE NOTICE 'TEST PASSED: df.start inserts nodes across the batch boundary';
END $$;

ROLLBACK;

RESET SESSION AUTHORIZATION;
SELECT 'TEST PASSED' AS result;