Idempotency keys for “create” endpoints

CREATE FUNCTION claim_idempotency_key(
    p_request_key   text,
    p_endpoint      text,
    p_fingerprint   text
)
RETURNS TABLE (did_claim boolean, response_code integer, response_body jsonb)
LANGUAGE plpgsql
AS $$
DECLARE
    v_existing idempotency_keys%ROWTYPE;
BEGIN
    -- Serialize concurrent requests that share this key (released at commit).
    PERFORM pg_advisory_xact_lock(hashtextextended(p_endpoint || ':' || p_request_key, 0));

    INSERT INTO idempotency_keys (request_key, endpoint, request_fingerprint)
    VALUES (p_request_key, p_endpoint, p_fingerprint)
    ON CONFLICT (request_key, endpoint) DO NOTHING;

    IF FOUND THEN
        RETURN QUERY SELECT true, NULL::integer, NULL::jsonb;
        RETURN;
    END IF;

    SELECT * INTO v_existing
    FROM idempotency_keys
    WHERE request_key = p_request_key AND endpoint = p_endpoint;

    IF v_existing.request_fingerprint <> p_fingerprint THEN
        RAISE EXCEPTION 'idempotency key reused with a different payload'
            USING ERRCODE = '23505';
    END IF;

    RETURN QUERY SELECT false, v_existing.response_code, v_existing.response_body;
END;
$$;

CREATE FUNCTION complete_idempotency_key(
    p_request_key text,
    p_endpoint    text,
    p_code        integer,
    p_body        jsonb,
    p_resource_id bigint
)
RETURNS void
LANGUAGE sql
AS $$
    UPDATE idempotency_keys
    SET status        = 'completed',
        response_code = p_code,
        response_body = p_body,
        resource_id   = p_resource_id,
        completed_at  = now()
    WHERE request_key = p_request_key AND endpoint = p_endpoint;
$$;

-- Runs inside a single transaction so the advisory lock spans the work.
CREATE FUNCTION create_order_idempotent(
    p_request_key text,
    p_fingerprint text,
    p_customer_id bigint,
    p_total_cents integer
)
RETURNS jsonb
LANGUAGE plpgsql
AS $$
DECLARE
    v_claim   record;
    v_order   record;
    v_body    jsonb;
BEGIN
    SELECT * INTO v_claim
    FROM claim_idempotency_key(p_request_key, 'POST /orders', p_fingerprint);

    IF NOT v_claim.did_claim THEN
        -- Retry of a completed request: replay the cached response verbatim.
        RETURN v_claim.response_body;
    END IF;

    INSERT INTO orders (customer_id, total_cents, status)
    VALUES (p_customer_id, p_total_cents, 'pending')
    RETURNING * INTO v_order;

    v_body := jsonb_build_object(
        'id',      v_order.id,
        'status',  v_order.status,
        'total',   v_order.total_cents
    );

    PERFORM complete_idempotency_key(
        p_request_key, 'POST /orders', 201, v_body, v_order.id
    );

    RETURN v_body;
END;
$$;