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

/* <begin connected objects> */
/*
This file is auto generated by pgrx.

The ordering of items is not stable, it is driven by a dependency graph.
*/
/* </end connected objects> */

/* <begin connected objects> */
-- src/lib.rs:146
CREATE SCHEMA IF NOT EXISTS df; /* pg_durable::df */
/* </end connected objects> */

/* <begin connected objects> */
-- src/lib.rs:153
-- requires:
--   df


-- Table to store function nodes (SQL steps, THEN chains, etc.)
CREATE TABLE IF NOT EXISTS df.nodes (
    id VARCHAR(8) PRIMARY KEY,
    instance_id VARCHAR(8),
    node_type TEXT NOT NULL,
    query TEXT,
    result_name TEXT,
    left_node VARCHAR(8),
    right_node VARCHAR(8),
    status TEXT DEFAULT 'pending',
    result JSONB,
    error TEXT,
    submitted_by REGROLE,
    database TEXT,
    created_at TIMESTAMPTZ DEFAULT pg_catalog.now(),
    updated_at TIMESTAMPTZ DEFAULT pg_catalog.now()
);

COMMENT ON COLUMN df.nodes.submitted_by IS
    'Effective role (current_user) at df.start() time - used for connection authentication and SQL execution';

-- Table to store function instances
CREATE TABLE IF NOT EXISTS df.instances (
    id VARCHAR(8) PRIMARY KEY,
    label TEXT,
    root_node VARCHAR(8) NOT NULL,
    status TEXT DEFAULT 'pending',
    submitted_by REGROLE NOT NULL,
    database TEXT,
    created_at TIMESTAMPTZ DEFAULT pg_catalog.now(),
    updated_at TIMESTAMPTZ DEFAULT pg_catalog.now(),
    completed_at TIMESTAMPTZ
);

COMMENT ON COLUMN df.instances.submitted_by IS
    'Effective role (current_user) at df.start() time - used for connection authentication and SQL execution';

-- Index for finding pending instances
CREATE INDEX IF NOT EXISTS idx_instances_status ON df.instances(status);

-- Index for finding nodes by instance
CREATE INDEX IF NOT EXISTS idx_nodes_instance ON df.nodes(instance_id);

-- Table to store workflow variables (captured at df.start())
-- Per-user scoping: each user has their own variable namespace.
CREATE TABLE IF NOT EXISTS df.vars (
    name TEXT NOT NULL,
    value TEXT,
    owner REGROLE NOT NULL DEFAULT pg_catalog.quote_ident(current_user)::pg_catalog.regrole,
    PRIMARY KEY (owner, name)
);

-- Sentinel table: the background worker writes its epoch_id here after
-- initialising.  If the extension is DROP-ed and re-CREATEd between
-- two poll ticks the epoch row disappears, so the worker detects the
-- recreation even though the extension is always "present" in pg_extension.
CREATE TABLE IF NOT EXISTS df._worker_epoch (
    epoch_id UUID PRIMARY KEY,
    started_at TIMESTAMPTZ DEFAULT pg_catalog.now(),
    last_seen_at TIMESTAMPTZ DEFAULT pg_catalog.now()
);

ALTER TABLE df.instances
    ADD CONSTRAINT instances_id_format_chk
        -- Operators (OPERATOR(pg_catalog.<op>)) and functions (e.g. pg_catalog.now)
        -- are schema-qualified throughout this install DDL so name resolution never
        -- depends on the session search_path -- closing the CVE-2018-1058 vector
        -- (a malicious schema shadowing `=`, `~`, etc.). Enforced by the pgspot CI
        -- gate (scripts/pgspot-gate.sh).
        CHECK (id OPERATOR(pg_catalog.~) '^[0-9a-f]{8}$') NOT VALID,
    ADD CONSTRAINT instances_root_node_format_chk
        CHECK (root_node OPERATOR(pg_catalog.~) '^[0-9a-f]{8}$') NOT VALID,
    ADD CONSTRAINT instances_status_chk
        CHECK (status OPERATOR(pg_catalog.=) ANY (ARRAY['pending', 'running', 'completed', 'failed', 'cancelled'])) NOT VALID,
    -- Supports the composite FK from df.nodes that ties node identity to the instance row.
    ADD CONSTRAINT instances_identity_key
        UNIQUE (id, submitted_by);

ALTER TABLE df.nodes
    ADD CONSTRAINT nodes_instance_id_present_chk
        CHECK (instance_id IS NOT NULL) NOT VALID,
    ADD CONSTRAINT nodes_submitted_by_present_chk
        CHECK (submitted_by IS NOT NULL) NOT VALID,
    ADD CONSTRAINT nodes_id_format_chk
        CHECK (id OPERATOR(pg_catalog.~) '^[0-9a-f]{8}$') NOT VALID,
    ADD CONSTRAINT nodes_instance_id_format_chk
        CHECK (instance_id OPERATOR(pg_catalog.~) '^[0-9a-f]{8}$') NOT VALID,
    ADD CONSTRAINT nodes_left_node_format_chk
        CHECK (left_node IS NULL OR left_node OPERATOR(pg_catalog.~) '^[0-9a-f]{8}$') NOT VALID,
    ADD CONSTRAINT nodes_right_node_format_chk
        CHECK (right_node IS NULL OR right_node OPERATOR(pg_catalog.~) '^[0-9a-f]{8}$') NOT VALID,
    ADD CONSTRAINT nodes_node_type_chk
        CHECK (node_type OPERATOR(pg_catalog.=) ANY (ARRAY['SQL', 'THEN', 'IF', 'JOIN', 'LOOP', 'BREAK', 'RACE', 'SLEEP', 'WAIT_SCHEDULE', 'HTTP', 'SIGNAL'])) NOT VALID,
    ADD CONSTRAINT nodes_result_name_chk
        CHECK (result_name IS NULL OR result_name OPERATOR(pg_catalog.~) '^[A-Za-z_][A-Za-z0-9_]*$') NOT VALID,
    ADD CONSTRAINT nodes_status_chk
        CHECK (status OPERATOR(pg_catalog.=) ANY (ARRAY['pending', 'running', 'completed', 'failed'])) NOT VALID,
    ADD CONSTRAINT nodes_result_status_chk
        CHECK (result IS NULL OR status OPERATOR(pg_catalog.=) ANY (ARRAY['completed', 'failed'])) NOT VALID,
    ADD CONSTRAINT nodes_structure_chk
        CHECK (
            CASE
                WHEN node_type OPERATOR(pg_catalog.=) ANY (ARRAY['SQL', 'SLEEP', 'WAIT_SCHEDULE', 'BREAK', 'HTTP', 'SIGNAL'])
                    THEN left_node IS NULL AND right_node IS NULL AND query IS NOT NULL
                WHEN node_type OPERATOR(pg_catalog.=) 'THEN'
                    THEN left_node IS NOT NULL AND right_node IS NOT NULL AND query IS NULL
                WHEN node_type OPERATOR(pg_catalog.=) 'IF'
                    THEN left_node IS NOT NULL AND right_node IS NOT NULL AND query IS NOT NULL
                WHEN node_type OPERATOR(pg_catalog.=) 'LOOP'
                    THEN left_node IS NOT NULL AND right_node IS NULL
                WHEN node_type OPERATOR(pg_catalog.=) 'JOIN'
                    THEN left_node IS NOT NULL AND right_node IS NOT NULL
                WHEN node_type OPERATOR(pg_catalog.=) 'RACE'
                    THEN left_node IS NOT NULL AND right_node IS NOT NULL AND query IS NULL
                ELSE FALSE
            END
        ) NOT VALID,
    ADD CONSTRAINT nodes_instance_node_key
        UNIQUE (instance_id, id);

ALTER TABLE df.nodes
    ADD CONSTRAINT nodes_instance_identity_fkey
        FOREIGN KEY (instance_id, submitted_by)
        REFERENCES df.instances (id, submitted_by)
        DEFERRABLE INITIALLY DEFERRED NOT VALID,
    ADD CONSTRAINT nodes_left_node_same_instance_fkey
        FOREIGN KEY (instance_id, left_node)
        REFERENCES df.nodes (instance_id, id)
        DEFERRABLE INITIALLY DEFERRED NOT VALID,
    ADD CONSTRAINT nodes_right_node_same_instance_fkey
        FOREIGN KEY (instance_id, right_node)
        REFERENCES df.nodes (instance_id, id)
        DEFERRABLE INITIALLY DEFERRED NOT VALID;

ALTER TABLE df.instances
    ADD CONSTRAINT instances_root_node_same_instance_fkey
        FOREIGN KEY (id, root_node)
        REFERENCES df.nodes (instance_id, id)
        DEFERRABLE INITIALLY DEFERRED NOT VALID;
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:924
-- pg_durable::dsl::cancel
CREATE  FUNCTION df."cancel"(
	"instance_id" TEXT, /* &str */
	"reason" TEXT DEFAULT 'Cancelled by user' /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'cancel_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/explain.rs:36
-- pg_durable::explain::explain
CREATE  FUNCTION df."explain"(
	"input" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'explain_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:48
-- pg_durable::dsl::debug_connection
CREATE  FUNCTION df."debug_connection"() RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'debug_connection_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:973
-- pg_durable::dsl::run
CREATE  FUNCTION df."run"(
	"instance_id" TEXT DEFAULT NULL /* core::option::Option<&str> */
) RETURNS TEXT /* alloc::string::String */

LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'run_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:38
-- pg_durable::dsl::version
CREATE  FUNCTION df."version"() RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'version_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:587
-- pg_durable::dsl::signal
CREATE  FUNCTION df."signal"(
	"instance_id" TEXT, /* &str */
	"signal_name" TEXT, /* &str */
	"signal_data" TEXT DEFAULT '{}' /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'signal_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:369
-- pg_durable::dsl::if
CREATE  FUNCTION df."if"(
	"condition" TEXT, /* &str */
	"then_branch" TEXT, /* &str */
	"else_branch" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT  
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'if_fn_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:414
-- pg_durable::dsl::join
CREATE  FUNCTION df."join"(
	"a" TEXT, /* &str */
	"b" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'join_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:1011
-- pg_durable::dsl::wait_for_completion
CREATE  FUNCTION df."wait_for_completion"(
	"instance_id" TEXT, /* &str */
	"timeout_seconds" INT DEFAULT 30 /* i32 */
) RETURNS TEXT /* core::result::Result<alloc::string::String, alloc::boxed::Box<dyn core::error::Error + core::marker::Send + core::marker::Sync>> */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'wait_for_completion_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/monitoring.rs:275
-- pg_durable::monitoring::metrics
CREATE  FUNCTION df."metrics"() RETURNS TABLE (
	"total_instances" bigint,  /* i64 */
	"running_instances" bigint,  /* i64 */
	"completed_instances" bigint,  /* i64 */
	"failed_instances" bigint,  /* i64 */
	"total_executions" bigint,  /* i64 */
	"total_events" bigint  /* i64 */
)
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'metrics_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:147
-- pg_durable::dsl::getvar
CREATE  FUNCTION df."getvar"(
	"name" TEXT /* &str */
) RETURNS TEXT /* core::option::Option<alloc::string::String> */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'getvar_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:124
-- pg_durable::dsl::setvar
CREATE  FUNCTION df."setvar"(
	"name" TEXT, /* &str */
	"value" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'setvar_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/monitoring.rs:323
-- pg_durable::monitoring::instance_nodes
CREATE  FUNCTION df."instance_nodes"(
	"instance_id_param" TEXT, /* &str */
	"last_n_executions" INT DEFAULT 5 /* i32 */
) RETURNS TABLE (
	"execution_id" bigint,  /* i64 */
	"node_id" TEXT,  /* alloc::string::String */
	"node_type" TEXT,  /* alloc::string::String */
	"query" TEXT,  /* core::option::Option<alloc::string::String> */
	"result_name" TEXT,  /* core::option::Option<alloc::string::String> */
	"left_node" TEXT,  /* core::option::Option<alloc::string::String> */
	"right_node" TEXT,  /* core::option::Option<alloc::string::String> */
	"status" TEXT,  /* core::option::Option<alloc::string::String> */
	"result" TEXT,  /* core::option::Option<alloc::string::String> */
	"updated_at" timestamp with time zone  /* core::option::Option<pgrx::datum::time_stamp_with_timezone::TimestampWithTimeZone> */
)
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'instance_nodes_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/types.rs:129
-- pg_durable::types::target_database
CREATE  FUNCTION df."target_database"() RETURNS TEXT /* alloc::string::String */
IMMUTABLE STRICT PARALLEL SAFE 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'target_database_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/lib.rs:574
-- requires:
--   df
--   target_database


-- Validate that CREATE EXTENSION is run in the correct database
-- The background worker connects to one specific database (determined by
-- the pg_durable.database GUC, defaults to "postgres").
-- The extension must be created in that database for workflows to execute.
DO $$
DECLARE
    current_db TEXT;
    target_db TEXT;
BEGIN
    -- Get the current database
    SELECT pg_catalog.current_database() INTO current_db;
    
    -- Get the target database that the background worker will connect to
    SELECT df.target_database() INTO target_db;
    
    IF current_db OPERATOR(pg_catalog.<>) target_db THEN
        RAISE EXCEPTION 'pg_durable extension must be created in database "%" (currently in "%"). The background worker only processes functions in the database specified by the pg_durable.database GUC (defaults to "postgres").', target_db, current_db
            USING HINT = 'Connect to the correct database and run: CREATE EXTENSION pg_durable;';
    END IF;
END $$;
/* </end connected objects> */

/* <begin connected objects> */
-- src/lib.rs:621
-- requires:
--   validate_database


-- The duroxide provider schema is created here so the extension owns it.
-- No IF NOT EXISTS: fails loudly if a duroxide schema already exists,
-- preventing adoption of a potentially attacker-crafted schema.
-- The background worker populates this schema at startup via ApplyAll.
CREATE SCHEMA duroxide;

/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:181
-- pg_durable::dsl::clearvars
CREATE  FUNCTION df."clearvars"() RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'clearvars_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:452
-- pg_durable::dsl::race
CREATE  FUNCTION df."race"(
	"a" TEXT, /* &str */
	"b" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'race_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:249
-- pg_durable::dsl::sleep
CREATE  FUNCTION df."sleep"(
	"seconds" bigint /* i64 */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'sleep_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:237
-- pg_durable::dsl::as
CREATE  FUNCTION df."as"(
	"fut" TEXT, /* &str */
	"name" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT  
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'as_named_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/monitoring.rs:116
-- pg_durable::monitoring::instance_info
CREATE  FUNCTION df."instance_info"(
	"instance_id" TEXT /* &str */
) RETURNS TABLE (
	"instance_id" TEXT,  /* alloc::string::String */
	"label" TEXT,  /* core::option::Option<alloc::string::String> */
	"function_name" TEXT,  /* alloc::string::String */
	"function_version" TEXT,  /* alloc::string::String */
	"current_execution_id" bigint,  /* i64 */
	"status" TEXT,  /* alloc::string::String */
	"output" TEXT  /* core::option::Option<alloc::string::String> */
)
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'instance_info_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:161
-- pg_durable::dsl::unsetvar
CREATE  FUNCTION df."unsetvar"(
	"name" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'unsetvar_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:206
-- pg_durable::dsl::sql
CREATE  FUNCTION df."sql"(
	"query" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'sql_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:219
-- pg_durable::dsl::seq
CREATE  FUNCTION df."seq"(
	"a" TEXT, /* &str */
	"b" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT  
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'then_fn_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:430
-- pg_durable::dsl::join3
CREATE  FUNCTION df."join3"(
	"a" TEXT, /* &str */
	"b" TEXT, /* &str */
	"c" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT  
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'join3_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:392
-- pg_durable::dsl::if_rows
CREATE  FUNCTION df."if_rows"(
	"result_name" TEXT, /* &str */
	"then_branch" TEXT, /* &str */
	"else_branch" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT  
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'if_rows_fn_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:265
-- pg_durable::dsl::wait_for_schedule
CREATE  FUNCTION df."wait_for_schedule"(
	"cron_expr" TEXT /* &str */
) RETURNS TEXT /* alloc::string::String */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'wait_for_schedule_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:983
-- pg_durable::dsl::result
CREATE  FUNCTION df."result"(
	"instance_id" TEXT /* &str */
) RETURNS TEXT /* core::option::Option<alloc::string::String> */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'result_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:353
-- pg_durable::dsl::break
CREATE  FUNCTION df."break"(
	"value" TEXT DEFAULT NULL /* core::option::Option<&str> */
) RETURNS TEXT /* alloc::string::String */
 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'break_fn_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:553
-- pg_durable::dsl::wait_for_signal
CREATE  FUNCTION df."wait_for_signal"(
	"name" TEXT, /* &str */
	"timeout_seconds" INT DEFAULT NULL /* core::option::Option<i32> */
) RETURNS TEXT /* alloc::string::String */

LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'wait_for_signal_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:312
-- pg_durable::dsl::loop
CREATE  FUNCTION df."loop"(
	"body" TEXT, /* &str */
	"condition" TEXT DEFAULT NULL /* core::option::Option<&str> */
) RETURNS TEXT /* alloc::string::String */
 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'loop_fn_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/lib.rs:652
-- requires:
--   dsl::then_fn
--   dsl::as_named
--   dsl::join
--   dsl::race
--   dsl::if_fn
--   dsl::loop_fn


-- Operator ~> for sequencing: a ~> b means "run a, then run b"
CREATE OPERATOR ~> (
    FUNCTION = df.seq,
    LEFTARG = text,
    RIGHTARG = text
);

-- Operator |=> for naming: fut |=> 'name' means "name this result as $name"
CREATE OR REPLACE FUNCTION df.as_op(fut text, name text) RETURNS text AS $$
    SELECT df.as(fut, name);
$$ LANGUAGE SQL IMMUTABLE SET search_path = pg_catalog, df, pg_temp;

CREATE OPERATOR |=> (
    FUNCTION = df.as_op,
    LEFTARG = text,
    RIGHTARG = text
);

-- Operator & for parallel join: a & b means "run a and b in parallel, wait for both"
CREATE OPERATOR & (
    FUNCTION = df.join,
    LEFTARG = text,
    RIGHTARG = text
);

-- Operator | for race: a | b means "run a and b in parallel, first wins"
CREATE OPERATOR | (
    FUNCTION = df.race,
    LEFTARG = text,
    RIGHTARG = text
);

-- Operators ?> and !> for if-then-else: cond ?> then_branch !> else_branch
-- We need helper functions to build the if node incrementally

-- Helper: cond ?> then creates a partial if (stores condition and then branch)
CREATE OR REPLACE FUNCTION df.if_then_op(condition text, then_branch text) RETURNS text AS $$
DECLARE
    cond_fut jsonb;
    then_fut jsonb;
    result_obj jsonb;
BEGIN
    -- Ensure both are durofuts
    cond_fut := df.ensure_durofut(condition)::jsonb;
    then_fut := df.ensure_durofut(then_branch)::jsonb;
    
    -- Return a special marker object for the partial if
    result_obj := jsonb_build_object(
        '_partial_if', true,
        'condition', cond_fut,
        'then_branch', then_fut
    );
    RETURN result_obj::text;
END;
$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, df, pg_temp;

-- Helper: partial_if !> else completes the if node
CREATE OR REPLACE FUNCTION df.if_else_op(partial_if text, else_branch text) RETURNS text AS $$
DECLARE
    partial jsonb;
    else_fut text;
    cond_text text;
    then_text text;
BEGIN
    partial := partial_if::jsonb;
    
    -- Check if it's a partial if
    IF partial->>'_partial_if' IS NULL THEN
        RAISE EXCEPTION 'Invalid if-then-else: left side of !> must be a ?> expression';
    END IF;
    
    cond_text := partial->'condition'::text;
    then_text := partial->'then_branch'::text;
    else_fut := df.ensure_durofut(else_branch);
    
    -- Now call the real df.if function
    RETURN df.if(cond_text, then_text, else_fut);
END;
$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, df, pg_temp;

-- Helper to ensure a value is a durofut (returns JSON string)
-- Rejects JSON with unknown node_type values.
-- NOTE: The valid node type list here must be kept in sync with
-- VALID_NODE_TYPES in src/types.rs (the Rust constant is the canonical source).
CREATE OR REPLACE FUNCTION df.ensure_durofut(val text) RETURNS text AS $$
DECLARE
    node_type_val text;
BEGIN
    -- Try to parse as JSON to check if it's already a durofut
    BEGIN
        node_type_val := (val::jsonb)->>'node_type';
        IF node_type_val IS NOT NULL THEN
            -- Has a node_type - validate it
            IF node_type_val NOT IN ('SQL', 'THEN', 'IF', 'JOIN', 'LOOP', 'BREAK', 'RACE', 'SLEEP', 'WAIT_SCHEDULE', 'HTTP', 'SIGNAL') THEN
                RAISE EXCEPTION 'Unknown node_type ''%''. Valid types: SQL, THEN, IF, JOIN, LOOP, BREAK, RACE, SLEEP, WAIT_SCHEDULE, HTTP, SIGNAL', node_type_val;
            END IF;
            RETURN val;
        END IF;
    EXCEPTION WHEN invalid_text_representation THEN
        -- Not valid JSON, treat as SQL
        NULL;
    WHEN raise_exception THEN
        -- Re-raise our validation error
        RAISE;
    WHEN OTHERS THEN
        -- Not valid JSON, treat as SQL
        NULL;
    END;
    
    -- It's plain SQL, wrap it
    RETURN df.sql(val);
END;
$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, df, pg_temp;

CREATE OPERATOR ?> (
    FUNCTION = df.if_then_op,
    LEFTARG = text,
    RIGHTARG = text
);

CREATE OPERATOR !> (
    FUNCTION = df.if_else_op,
    LEFTARG = text,
    RIGHTARG = text
);

-- Operator @> for loop: @> body means "repeat body forever"
-- This is a PREFIX operator with lowest precedence
CREATE OR REPLACE FUNCTION df.loop_prefix_op(body text) RETURNS text AS $$
    SELECT df.loop(body);
$$ LANGUAGE SQL IMMUTABLE SET search_path = pg_catalog, df, pg_temp;

CREATE OPERATOR @> (
    FUNCTION = df.loop_prefix_op,
    RIGHTARG = text
);
/* </end connected objects> */

/* <begin connected objects> */
-- src/monitoring.rs:195
-- pg_durable::monitoring::instance_executions
CREATE  FUNCTION df."instance_executions"(
	"instance_id" TEXT, /* &str */
	"limit_count" INT DEFAULT 5 /* i32 */
) RETURNS TABLE (
	"execution_id" bigint,  /* i64 */
	"status" TEXT,  /* alloc::string::String */
	"event_count" bigint,  /* i64 */
	"duration_ms" bigint,  /* i64 */
	"output" TEXT  /* core::option::Option<alloc::string::String> */
)
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'instance_executions_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/monitoring.rs:18
-- pg_durable::monitoring::list_instances
CREATE  FUNCTION df."list_instances"(
	"status_filter" TEXT DEFAULT NULL, /* core::option::Option<&str> */
	"limit_count" INT DEFAULT 100 /* i32 */
) RETURNS TABLE (
	"instance_id" TEXT,  /* alloc::string::String */
	"label" TEXT,  /* core::option::Option<alloc::string::String> */
	"function_name" TEXT,  /* alloc::string::String */
	"status" TEXT,  /* alloc::string::String */
	"execution_count" bigint,  /* i64 */
	"output" TEXT  /* core::option::Option<alloc::string::String> */
)

LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'list_instances_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:962
-- pg_durable::dsl::status
CREATE  FUNCTION df."status"(
	"instance_id" TEXT /* &str */
) RETURNS TEXT /* core::option::Option<alloc::string::String> */
STRICT 
LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'status_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:478
-- pg_durable::dsl::http
CREATE  FUNCTION df."http"(
	"url" TEXT, /* &str */
	"method" TEXT DEFAULT 'POST', /* &str */
	"body" TEXT DEFAULT NULL, /* core::option::Option<&str> */
	"headers" jsonb DEFAULT NULL, /* core::option::Option<pgrx::datum::json::JsonB> */
	"timeout_seconds" INT DEFAULT 30 /* i32 */
) RETURNS TEXT /* alloc::string::String */

LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'http_wrapper';
/* </end connected objects> */

/* <begin connected objects> */
-- src/lib.rs:303
-- requires:
--   create_tables
--   dsl::http


-- Enable RLS on df.instances (no FORCE — superuser/table-owner bypasses RLS)
ALTER TABLE df.instances ENABLE ROW LEVEL SECURITY;

CREATE POLICY instances_user_isolation ON df.instances
    FOR ALL
    USING (submitted_by OPERATOR(pg_catalog.=) pg_catalog.quote_ident(current_user)::pg_catalog.regrole)
    WITH CHECK (submitted_by OPERATOR(pg_catalog.=) pg_catalog.quote_ident(current_user)::pg_catalog.regrole);

-- Enable RLS on df.nodes
ALTER TABLE df.nodes ENABLE ROW LEVEL SECURITY;

CREATE POLICY nodes_user_isolation ON df.nodes
    FOR ALL
    USING (submitted_by OPERATOR(pg_catalog.=) pg_catalog.quote_ident(current_user)::pg_catalog.regrole)
    WITH CHECK (submitted_by OPERATOR(pg_catalog.=) pg_catalog.quote_ident(current_user)::pg_catalog.regrole);

-- Enable RLS on df.vars (per-user variable isolation)
ALTER TABLE df.vars ENABLE ROW LEVEL SECURITY;

CREATE POLICY vars_user_isolation ON df.vars
    FOR ALL
    USING (owner OPERATOR(pg_catalog.=) pg_catalog.quote_ident(current_user)::pg_catalog.regrole)
    WITH CHECK (owner OPERATOR(pg_catalog.=) pg_catalog.quote_ident(current_user)::pg_catalog.regrole);

-- No automatic PUBLIC grants. Admins must explicitly grant privileges
-- to application roles after CREATE EXTENSION.
-- Use df.grant_usage('role_name') (recommended) or see USER_GUIDE.md
-- "Privilege Grants" for the equivalent manual GRANT statements.

-- Helper: grant all required df privileges to a role in one call.
-- Authorization model: This function is SECURITY INVOKER and EXECUTE is
-- revoked from PUBLIC, so only roles explicitly granted EXECUTE (via
-- with_grant => true or a direct superuser GRANT) can call it. The inner
-- GRANT statements run as the caller, so the caller must also hold the
-- underlying privileges WITH GRANT OPTION (automatically true for
-- superusers; for delegated admins, df.grant_usage(..., with_grant => true)
-- grants all privileges WITH GRANT OPTION).
--
-- This function is purely additive — it never issues REVOKE.  To downgrade
-- a role's privileges, call df.revoke_usage() first, then df.grant_usage()
-- with the desired options.
--
-- include_http controls whether the role is granted EXECUTE on df.http().
-- Default is false: HTTP access is opt-in because df.http() makes outbound
-- network requests and requires explicit administrator approval.
--
-- with_grant controls whether the target role receives privileges WITH GRANT
-- OPTION and can itself call df.grant_usage() / df.revoke_usage() to manage
-- other roles' access.  Authorization is enforced by PostgreSQL's native
-- WITH GRANT OPTION mechanism: the caller must hold each underlying
-- privilege WITH GRANT OPTION, which is automatically true for superusers
-- and for delegated admins granted via with_grant => true.
--
-- MAINTENANCE: when adding a new df.* function, add it to the func_sigs
-- array below.  Functions NOT in this list are deny-by-default.
CREATE OR REPLACE FUNCTION df.grant_usage(
    p_role TEXT,
    include_http boolean DEFAULT false,
    with_grant boolean DEFAULT false
)
RETURNS VOID
LANGUAGE plpgsql
SET search_path = pg_catalog, df, pg_temp
AS $fn$
DECLARE
    grant_opt TEXT := '';
    func_sig TEXT;
    -- Explicit list of df.* functions to grant.  Sensitive functions
    -- (df.http, df.grant_usage, df.revoke_usage) are excluded from this
    -- list and granted conditionally below.
    func_sigs TEXT[] := ARRAY[
        -- DSL functions
        'df.sql(text)',
        'df.seq(text, text)',
        'df.as(text, text)',
        'df.sleep(bigint)',
        'df.wait_for_schedule(text)',
        'df.loop(text, text)',
        'df.break(text)',
        'df.if(text, text, text)',
        'df.if_rows(text, text, text)',
        'df.join(text, text)',
        'df.join3(text, text, text)',
        'df.race(text, text)',
        'df.wait_for_signal(text, integer)',
        'df.signal(text, text, text)',
        'df.start(text, text, text)',
        'df.setvar(text, text)',
        'df.getvar(text)',
        'df.unsetvar(text)',
        'df.clearvars()',
        -- Monitoring functions
        'df.status(text)',
        'df.result(text)',
        'df.cancel(text, text)',
        'df.wait_for_completion(text, integer)',
        'df.run(text)',
        'df.list_instances(text, integer)',
        'df.instance_info(text)',
        'df.instance_nodes(text, integer)',
        'df.instance_executions(text, integer)',
        'df.metrics()',
        -- Internal helpers (operators, version, etc.)
        'df.as_op(text, text)',
        'df.if_then_op(text, text)',
        'df.if_else_op(text, text)',
        'df.ensure_durofut(text)',
        'df.loop_prefix_op(text)',
        'df.version()',
        'df.debug_connection()',
        'df.explain(text)',
        'df.target_database()'
    ];
BEGIN
    -- Validate the role exists
    IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = p_role) THEN
        RAISE EXCEPTION 'role "%" does not exist', p_role;
    END IF;

    IF with_grant THEN
        grant_opt := ' WITH GRANT OPTION';
    END IF;

    -- Schema access
    EXECUTE format('GRANT USAGE ON SCHEMA df TO %I', p_role) || grant_opt;

    -- Grant EXECUTE on each standard function explicitly.
    FOREACH func_sig IN ARRAY func_sigs LOOP
        EXECUTE format('GRANT EXECUTE ON FUNCTION %s TO %I', func_sig, p_role) || grant_opt;
    END LOOP;

    -- df.http() — opt-in because it makes outbound network requests.
    IF include_http THEN
        EXECUTE format('GRANT EXECUTE ON FUNCTION df.http(text, text, text, jsonb, integer) TO %I', p_role) || grant_opt;
    END IF;

    -- Admin helpers — only for delegated administrators.
    IF with_grant THEN
        EXECUTE format('GRANT EXECUTE ON FUNCTION df.grant_usage(text, boolean, boolean) TO %I', p_role) || grant_opt;
        EXECUTE format('GRANT EXECUTE ON FUNCTION df.revoke_usage(text) TO %I', p_role) || grant_opt;
    END IF;

    -- Table privileges
    EXECUTE format('GRANT SELECT ON df.instances TO %I', p_role) || grant_opt;
    EXECUTE format('GRANT UPDATE (status, updated_at) ON df.instances TO %I', p_role) || grant_opt;
    EXECUTE format('GRANT SELECT ON df.nodes TO %I', p_role) || grant_opt;
    EXECUTE format('GRANT INSERT (id, label, root_node, submitted_by, database) ON df.instances TO %I', p_role) || grant_opt;
    EXECUTE format('GRANT INSERT (id, instance_id, node_type, query, result_name, left_node, right_node, submitted_by, database) ON df.nodes TO %I', p_role) || grant_opt;
    EXECUTE format('GRANT SELECT, INSERT, UPDATE, DELETE ON df.vars TO %I', p_role) || grant_opt;

    RAISE NOTICE 'pg_durable: granted df usage privileges to "%"', p_role;
END;
$fn$;

-- Revoke all df privileges previously granted by df.grant_usage().
-- Authorization: same model as df.grant_usage() — EXECUTE is revoked from
-- PUBLIC, caller must hold the underlying privileges to revoke them.
-- Safety: format(%I) quotes identifiers to prevent SQL injection. Additionally,
-- this is SECURITY INVOKER so it cannot escalate beyond the caller's privileges.
CREATE OR REPLACE FUNCTION df.revoke_usage(p_role TEXT)
RETURNS VOID
LANGUAGE plpgsql
SET search_path = pg_catalog, df, pg_temp
AS $fn$
DECLARE
    func_oid oid;
BEGIN
    -- Validate the role exists
    IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = p_role) THEN
        RAISE EXCEPTION 'role "%" does not exist', p_role;
    END IF;

    -- Prevent accidentally revoking your own access.  pg_has_role checks
    -- both direct identity (current_user = p_role) and inherited membership
    -- (current_user is a member of p_role), so revoking a parent role that
    -- the caller depends on is also caught.
    -- Superusers are exempt: pg_has_role returns true for all roles when the
    -- caller is a superuser, and superusers can always re-grant themselves.
    IF NOT EXISTS (
        SELECT 1
        FROM pg_roles
        WHERE rolname = current_user
          AND rolsuper
    )
       AND pg_has_role(current_user, p_role, 'MEMBER') THEN
        RAISE EXCEPTION 'cannot revoke df privileges from "%" because the current role ("%") is a member of it — use a different administrator', p_role, current_user;
    END IF;

    -- CASCADE: if the target role granted sub-grants (via WITH GRANT OPTION),
    -- CASCADE ensures those dependent privileges are also revoked.
    -- Column-level revokes must match the column-level grants from grant_usage().
    EXECUTE format('REVOKE SELECT, INSERT, UPDATE, DELETE ON df.vars FROM %I CASCADE', p_role);
    EXECUTE format('REVOKE INSERT (id, instance_id, node_type, query, result_name, left_node, right_node, submitted_by, database) ON df.nodes FROM %I CASCADE', p_role);
    EXECUTE format('REVOKE SELECT ON df.nodes FROM %I CASCADE', p_role);
    EXECUTE format('REVOKE INSERT (id, label, root_node, submitted_by, database) ON df.instances FROM %I CASCADE', p_role);
    EXECUTE format('REVOKE UPDATE (status, updated_at) ON df.instances FROM %I CASCADE', p_role);
    EXECUTE format('REVOKE SELECT ON df.instances FROM %I CASCADE', p_role);

    -- Revoke EXECUTE per-function rather than using the blanket
    -- REVOKE ON ALL FUNCTIONS.  A delegated admin may lack privilege on
    -- some functions (e.g. df.http); per-function revokes let us skip those.
    FOR func_oid IN
        SELECT p.oid FROM pg_proc p
        JOIN pg_namespace n ON p.pronamespace = n.oid
        WHERE n.nspname = 'df'
    LOOP
        BEGIN
            EXECUTE format('REVOKE EXECUTE ON FUNCTION %s FROM %I CASCADE', func_oid::regprocedure, p_role);
        EXCEPTION WHEN insufficient_privilege THEN
            NULL;
        END;
    END LOOP;

    EXECUTE format('REVOKE USAGE ON SCHEMA df FROM %I CASCADE', p_role);

    RAISE NOTICE 'pg_durable: revoked df usage privileges granted by "%" from "%"', current_user, p_role;
END;
$fn$;

-- Validate that the worker role is a superuser.
-- The background worker must bypass RLS to manage all users' instances/nodes.
-- If the worker role is not a superuser, workflows will silently fail because
-- RLS will filter out rows the worker needs to read/update.
DO $$
DECLARE
    wrole TEXT;
    is_super BOOLEAN;
BEGIN
    wrole := pg_catalog.current_setting('pg_durable.worker_role', true);
    IF wrole IS NULL OR wrole OPERATOR(pg_catalog.=) '' THEN
        wrole := 'azuresu';
    END IF;

    SELECT rolsuper INTO is_super FROM pg_catalog.pg_roles WHERE rolname OPERATOR(pg_catalog.=) wrole;
    IF is_super IS NULL THEN
        RAISE WARNING 'pg_durable: worker role "%" does not exist. The background worker will not be able to process workflows. Create the role as a superuser before using pg_durable.', wrole;
    ELSIF NOT is_super THEN
        RAISE WARNING 'pg_durable: worker role "%" is not a superuser. The background worker must be a superuser to bypass RLS and manage all users'' instances. Grant superuser or BYPASSRLS to this role.', wrole;
    END IF;
END $$;

-- df.http() carries network-access implications and must be opt-in.
-- PostgreSQL grants EXECUTE to PUBLIC by default when functions are created;
-- revoke that so only roles explicitly granted access (via
-- df.grant_usage(role, include_http => true) or a direct GRANT) can make
-- HTTP requests.
REVOKE EXECUTE ON FUNCTION df.http(text, text, text, jsonb, integer) FROM PUBLIC;

-- df.grant_usage() and df.revoke_usage() are admin-only helpers.
-- Revoke PUBLIC's default EXECUTE privilege so that only roles explicitly
-- granted access (via with_grant => true or a direct superuser GRANT) can
-- manage other roles' df privileges.
REVOKE EXECUTE ON FUNCTION df.grant_usage(text, boolean, boolean) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION df.revoke_usage(text) FROM PUBLIC;
/* </end connected objects> */

/* <begin connected objects> */
-- src/dsl.rs:630
-- pg_durable::dsl::start
CREATE  FUNCTION df."start"(
	"fut" TEXT, /* &str */
	"label" TEXT DEFAULT NULL, /* core::option::Option<&str> */
	"database" TEXT DEFAULT NULL /* core::option::Option<&str> */
) RETURNS TEXT /* alloc::string::String */

LANGUAGE c /* Rust */
AS 'MODULE_PATHNAME', 'start_wrapper';
/* </end connected objects> */