pgsql-broker/pkg/broker/install/sql/migrations/0011_broker_add_job.sql

-- broker.broker_add_job
-- Adds a new job (optionally with dependencies and an idempotency key) and
-- sends a wake-only NOTIFY -- the payload carries only the queue number
-- (job id kept solely for logging); workers re-claim via broker_get rather
-- than executing the notified row directly, so a notification can never
-- hand a job to a worker before it's actually claimable.

CREATE OR REPLACE FUNCTION broker.broker_add_job(
    p_job_name TEXT,
    p_execute_str TEXT,
    p_job_queue INTEGER DEFAULT 1,
    p_job_priority INTEGER DEFAULT 0,
    p_job_language TEXT DEFAULT 'sql',
    p_run_as TEXT DEFAULT NULL,
    p_schedule_id BIGINT DEFAULT NULL,
    p_depends_on_job_ids BIGINT[] DEFAULT NULL,
    p_idempotency_key TEXT DEFAULT NULL,
    p_max_attempts INTEGER DEFAULT 1,
    OUT p_retval INTEGER,
    OUT p_errmsg TEXT,
    OUT p_job_id BIGINT
)
RETURNS RECORD
LANGUAGE plpgsql
AS $$
DECLARE
    v_notification_payload JSON;
    v_tenant_id TEXT;
    v_dep_id BIGINT;
    v_cycle_exists BOOLEAN;
BEGIN
    p_retval := 0;
    p_errmsg := '';
    p_job_id := NULL;

    IF p_job_name IS NULL OR p_job_name = '' THEN
        p_retval := 1;
        p_errmsg := 'Job name is required';
        RETURN;
    END IF;

    IF p_execute_str IS NULL OR p_execute_str = '' THEN
        p_retval := 2;
        p_errmsg := 'Execute string is required';
        RETURN;
    END IF;

    IF p_job_queue IS NULL OR p_job_queue <= 0 THEN
        p_retval := 3;
        p_errmsg := 'Invalid job queue number';
        RETURN;
    END IF;

    IF p_max_attempts IS NULL OR p_max_attempts <= 0 THEN
        p_max_attempts := 1;
    END IF;

    -- Falls back to 'default' when the caller never called broker_set_tenant,
    -- so single-tenant use (and the RLS WITH CHECK on insert) keeps working.
    v_tenant_id := COALESCE(NULLIF(current_setting('broker.tenant_id', true), ''), 'default');

    IF p_idempotency_key IS NOT NULL THEN
        SELECT id_broker_jobs INTO p_job_id
        FROM broker.broker_jobs
        WHERE job_queue = p_job_queue
          AND idempotency_key = p_idempotency_key
          AND tenant_id = v_tenant_id;

        IF FOUND THEN
            p_errmsg := 'Job with this idempotency key already exists; returning existing job id';
            RETURN;
        END IF;
    END IF;

    INSERT INTO broker.broker_jobs (
        job_name, job_priority, job_queue, job_language, execute_str, run_as,
        rid_broker_schedule, tenant_id, max_attempts, idempotency_key, complete_status
    ) VALUES (
        p_job_name, p_job_priority, p_job_queue, p_job_language, p_execute_str, p_run_as,
        p_schedule_id, v_tenant_id, p_max_attempts, p_idempotency_key, 0
    )
    RETURNING id_broker_jobs INTO p_job_id;

    IF p_depends_on_job_ids IS NOT NULL THEN
        FOREACH v_dep_id IN ARRAY p_depends_on_job_ids LOOP
            IF v_dep_id IS NULL THEN
                CONTINUE;
            END IF;

            IF v_dep_id = p_job_id THEN
                p_retval := 20;
                p_errmsg := 'Invalid dependency: a job cannot depend on itself';
                RETURN;
            END IF;

            SELECT EXISTS (
                SELECT 1 FROM broker.broker_job_dependency
                WHERE job_id = v_dep_id AND depends_on_job_id = p_job_id
            ) INTO v_cycle_exists;

            IF v_cycle_exists THEN
                p_retval := 21;
                p_errmsg := format('Invalid dependency: job %s already depends on %s (would create a cycle)', v_dep_id, p_job_id);
                RETURN;
            END IF;

            INSERT INTO broker.broker_job_dependency (job_id, depends_on_job_id, tenant_id)
            VALUES (p_job_id, v_dep_id, v_tenant_id)
            ON CONFLICT (job_id, depends_on_job_id) DO NOTHING;
        END LOOP;
    END IF;

    v_notification_payload := json_build_object(
        'queue', p_job_queue,
        'job_id', p_job_id
    );

    PERFORM pg_notify('broker.event', v_notification_payload::text);

EXCEPTION
    WHEN unique_violation THEN
        -- Concurrent insert raced us to the same idempotency key.
        p_retval := 0;
        p_errmsg := 'Job with this idempotency key already exists; returning existing job id';
        SELECT id_broker_jobs INTO p_job_id
        FROM broker.broker_jobs
        WHERE job_queue = p_job_queue
          AND idempotency_key = p_idempotency_key
          AND tenant_id = v_tenant_id;
    WHEN OTHERS THEN
        p_retval := 99;
        p_errmsg := SQLERRM;
        RAISE WARNING 'broker_add_job error: %', SQLERRM;
END;
$$;

COMMENT ON FUNCTION broker.broker_add_job IS 'Adds a job (with optional dependencies/idempotency key) and sends a wake-only NOTIFY';