member-console/internal/db/migrations/00005_doc41_conferral_functions.sql

-- +goose Up

-- Doc 41 §9 step 8: the conferral primitive and its enclosure. Five
-- SECURITY DEFINER functions become the sole legal writers of
-- core.pool_provisions, core.pool_provision_ladders, and
-- core.pool_provision_transitions; core_writer loses direct DML on those
-- three tables and gains EXECUTE on exactly these five members.
--
-- Named error SQLSTATEs (stable contract for the Go layer, design D3):
--   CF001 conferral_precedes_incumbent
--   CF002 conferral_shape_diverged
--   CF003 conferral_shape_collision
--   CF004 conferral_already_ended
--
-- Callers never choose position behavior: everything beyond (pool, product,
-- source, bounds, attribution) is derived from core.product_conferral_shapes.
-- Materialization stays application-side (Doc 41 §8, below-the-line).

-- +goose StatementBegin
CREATE FUNCTION core.confer(
    p_pool_id          UUID,
    p_product_id       UUID,
    p_subscription_id  UUID DEFAULT NULL,
    p_purchase_id      UUID DEFAULT NULL,
    p_grant_id         UUID DEFAULT NULL,
    p_quantity         INTEGER DEFAULT 1,
    p_activated_at     TIMESTAMPTZ DEFAULT NULL,
    p_ended_at         TIMESTAMPTZ DEFAULT NULL,
    p_actor_type       VARCHAR DEFAULT 'system',
    p_actor_id         UUID DEFAULT NULL,
    p_reason           TEXT DEFAULT NULL
) RETURNS TABLE (provision_id UUID, outcome VARCHAR)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = core, pg_temp
AS $$
DECLARE
    v_source_count      INTEGER;
    v_set_id            UUID;
    v_lifecycle         VARCHAR;
    v_ladder_count      BIGINT;
    v_grant_reason      VARCHAR;
    v_extends_grant_id  UUID;
    v_incumbent         RECORD;
    v_same              RECORD;
    v_missing           BIGINT;
    v_target            RECORD;
    v_junction          RECORD;
    v_from_rank         INTEGER;
    v_new_rank          INTEGER;
    v_type              VARCHAR;
    v_new_provision_id  UUID;
    v_shape             RECORD;
    v_new_junction_id   UUID;
    v_from_ranks        JSONB := '{}'::jsonb;
    v_activated_at      TIMESTAMPTZ;
BEGIN
    -- Step 1: sanity. Exactly one source; no forward-dated activation
    -- (future intent lives in subscription_scheduled_changes).
    v_source_count := (p_subscription_id IS NOT NULL)::int
                    + (p_purchase_id IS NOT NULL)::int
                    + (p_grant_id IS NOT NULL)::int;
    IF v_source_count <> 1 THEN
        RAISE EXCEPTION 'confer: exactly one of subscription/purchase/grant must be supplied (got %)', v_source_count;
    END IF;
    -- NULL means "now", supplied by the database — callers never fabricate a
    -- wall-clock "now" (a Go time.Now() is always ahead of the frozen
    -- transaction_timestamp(), so comparing against NOW() rejects every
    -- honest call). Explicit values are guarded against execution time.
    v_activated_at := COALESCE(p_activated_at, NOW());
    IF p_activated_at IS NOT NULL AND p_activated_at > clock_timestamp() THEN
        RAISE EXCEPTION 'confer: p_activated_at % is in the future; scheduled changes are a separate mechanism', p_activated_at;
    END IF;
    IF p_quantity IS NULL OR p_quantity < 1 THEN
        RAISE EXCEPTION 'confer: quantity must be a positive integer (got %)', p_quantity;
    END IF;

    -- Step 2: serialize concurrent conferrals on the pool.
    PERFORM 1 FROM core.resource_pools rp WHERE rp.pool_id = p_pool_id FOR UPDATE;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'confer: pool % does not exist', p_pool_id;
    END IF;

    -- Step 3: availability, read only from the conferral-shape view.
    SELECT DISTINCT s.entitlement_set_id, s.lifecycle_status, s.ladder_count
    INTO v_set_id, v_lifecycle, v_ladder_count
    FROM core.product_conferral_shapes s
    WHERE s.product_id = p_product_id;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'confer: product % does not exist', p_product_id;
    END IF;
    IF v_set_id IS NULL THEN
        RAISE EXCEPTION 'confer: product % has no entitlement set and confers nothing', p_product_id;
    END IF;

    IF p_grant_id IS NOT NULL THEN
        SELECT g.grant_reason, g.extends_grant_id INTO v_grant_reason, v_extends_grant_id
        FROM core.grants g WHERE g.grant_id = p_grant_id;
        IF NOT FOUND THEN
            RAISE EXCEPTION 'confer: grant % does not exist', p_grant_id;
        END IF;
    END IF;

    IF v_lifecycle = 'draft' THEN
        RAISE EXCEPTION 'confer: product % is draft and not conferrable', p_product_id;
    ELSIF v_lifecycle = 'retired' THEN
        -- Re-conferral of a retired product is continuity, not new sale:
        -- permitted only when re-enacting a standing source. Default grants
        -- use the source-class predicate (reapply mints fresh grant rows).
        IF v_grant_reason = 'default' THEN
            PERFORM 1
            FROM core.pool_provisions pp
            JOIN core.grants g2 ON g2.grant_id = pp.grant_id
            WHERE pp.pool_id = p_pool_id
              AND pp.product_id = p_product_id
              AND g2.grant_reason = 'default';
        ELSE
            PERFORM 1
            FROM core.pool_provisions pp
            WHERE pp.pool_id = p_pool_id
              AND pp.product_id = p_product_id
              AND ((p_subscription_id IS NOT NULL AND pp.subscription_id = p_subscription_id)
                OR (p_purchase_id     IS NOT NULL AND pp.purchase_id     = p_purchase_id)
                OR (p_grant_id        IS NOT NULL AND pp.grant_id        = p_grant_id));
        END IF;
        IF NOT FOUND THEN
            RAISE EXCEPTION 'confer: product % is retired; fresh conferral is intake, not continuity', p_product_id;
        END IF;
    END IF;

    -- Step 4a: source-scoped idempotence. The same source already
    -- delivering the same product (active OR suspended) is a retry — no-op.
    SELECT pp.provision_id AS id, pp.product_id AS pid
    INTO v_incumbent
    FROM core.pool_provisions pp
    WHERE pp.pool_id = p_pool_id
      AND pp.status <> 'ended'
      AND ((p_subscription_id IS NOT NULL AND pp.subscription_id = p_subscription_id AND pp.product_id = p_product_id)
        OR (p_purchase_id     IS NOT NULL AND pp.purchase_id     = p_purchase_id)
        OR (p_grant_id        IS NOT NULL AND pp.grant_id        = p_grant_id))
    LIMIT 1;
    IF v_incumbent.id IS NOT NULL THEN
        IF v_incumbent.pid = p_product_id THEN
            RETURN QUERY SELECT v_incumbent.id, 'noop'::VARCHAR;
            RETURN;
        END IF;
        -- Same source, different product: the old delivery is this source's
        -- own affair (reconcile diffs end it via end_conferral); position
        -- supersession below handles any contested rungs it holds.
    END IF;

    -- Step 4b: same-product incumbents from other sources.
    FOR v_same IN
        SELECT pp.provision_id AS id, pp.grant_id AS gid,
               pp.subscription_id AS sid, pp.purchase_id AS puid
        FROM core.pool_provisions pp
        WHERE pp.pool_id = p_pool_id
          AND pp.product_id = p_product_id
          AND pp.status <> 'ended'
    LOOP
        -- Shape divergence: the incumbent occupies only part of the
        -- product's CURRENT shape (catalog evolved after conferral).
        SELECT COUNT(*) INTO v_missing
        FROM core.product_conferral_shapes s
        WHERE s.product_id = p_product_id
          AND s.plan_ladder_id IS NOT NULL
          AND NOT EXISTS (
              SELECT 1 FROM core.pool_provision_ladders l
              WHERE l.provision_id = v_same.id
                AND l.plan_ladder_id = s.plan_ladder_id
                AND l.status <> 'ended'
          );
        IF v_missing > 0 THEN
            RAISE EXCEPTION USING
                ERRCODE = 'CF002',
                MESSAGE = format('confer: provision %s of product %s does not match the product''s current shape; run align_conferral_shape first', v_same.id, p_product_id);
        END IF;

        -- Decree-duplication no-op (position-bearing, grant-over-grant
        -- only): a second unrelated grant of a product whose position the
        -- pool already holds changes nothing live — unless it extends the
        -- incumbent's grant, which is supersession (extend-as-replace).
        IF p_grant_id IS NOT NULL
           AND v_ladder_count > 0
           AND v_same.gid IS NOT NULL
           AND (v_extends_grant_id IS NULL OR v_extends_grant_id <> v_same.gid) THEN
            RETURN QUERY SELECT v_same.id, 'noop'::VARCHAR;
            RETURN;
        END IF;
    END LOOP;

    -- Steps 5+6: supersession and creation.
    v_new_provision_id := uuidv7();

    IF v_ladder_count > 0 THEN
        -- Capture pre-supersession incumbent ranks per contested ladder
        -- (from_rank inputs for the new position's transition rows) before
        -- anything is ended. Prefer the active incumbent over suspended.
        SELECT COALESCE(jsonb_object_agg(x.lid::text, x.r), '{}'::jsonb) INTO v_from_ranks
        FROM (
            SELECT DISTINCT ON (l.plan_ladder_id) l.plan_ladder_id AS lid, t.rank AS r
            FROM core.pool_provision_ladders l
            LEFT JOIN core.plan_ladder_tiers t
                   ON t.plan_ladder_id = l.plan_ladder_id AND t.product_id = l.product_id
            WHERE l.pool_id = p_pool_id
              AND l.status IN ('active', 'suspended')
              AND l.plan_ladder_id IN (
                  SELECT s.plan_ladder_id FROM core.product_conferral_shapes s
                  WHERE s.product_id = p_product_id AND s.plan_ladder_id IS NOT NULL
              )
            ORDER BY l.plan_ladder_id, (l.status = 'active') DESC, l.activated_at DESC
        ) x;

        -- Every live-or-suspended incumbent on any contested rung is
        -- superseded — suspended incumbents count as occupying (a past-due
        -- subscription recovering later must not detonate the exclusion
        -- constraint inside the sync trigger).
        FOR v_target IN
            SELECT DISTINCT pp.provision_id AS id, pp.activated_at AS act
            FROM core.pool_provision_ladders l
            JOIN core.pool_provisions pp ON pp.provision_id = l.provision_id
            WHERE l.pool_id = p_pool_id
              AND l.status IN ('active', 'suspended')
              AND l.plan_ladder_id IN (
                  SELECT s.plan_ladder_id FROM core.product_conferral_shapes s
                  WHERE s.product_id = p_product_id AND s.plan_ladder_id IS NOT NULL
              )
        LOOP
            -- Backdating never reaches under an incumbent it supersedes.
            IF v_activated_at < v_target.act THEN
                RAISE EXCEPTION USING
                    ERRCODE = 'CF001',
                    MESSAGE = format('confer: activation %s precedes incumbent provision %s activated %s', v_activated_at, v_target.id, v_target.act);
            END IF;

            -- Whole-bundle end: one 'end' transition per rung the
            -- superseded provision held, uncontested rungs included.
            FOR v_junction IN
                SELECT l.provision_ladder_id AS jid, l.plan_ladder_id AS lid, l.product_id AS pid
                FROM core.pool_provision_ladders l
                WHERE l.provision_id = v_target.id
                  AND l.status <> 'ended'
            LOOP
                SELECT t.rank INTO v_from_rank
                FROM core.plan_ladder_tiers t
                WHERE t.plan_ladder_id = v_junction.lid AND t.product_id = v_junction.pid;

                INSERT INTO core.pool_provision_transitions
                    (pool_id, provision_id, plan_ladder_id, from_rank, to_rank,
                     transition_type, actor_type, actor_id, reason, effective_at, provision_ladder_id)
                VALUES
                    (p_pool_id, v_target.id, v_junction.lid, v_from_rank, NULL,
                     'end', p_actor_type, p_actor_id, p_reason, v_activated_at, v_junction.jid);
            END LOOP;

            UPDATE core.pool_provisions pp
            SET status = 'ended', ended_at = v_activated_at
            WHERE pp.provision_id = v_target.id;
            -- (sync trigger ends its non-ended junction rows)
        END LOOP;
    END IF;

    INSERT INTO core.pool_provisions
        (provision_id, pool_id, subscription_id, purchase_id, grant_id,
         quantity, status, activated_at, ended_at, entitlement_set_id, product_id)
    VALUES
        (v_new_provision_id, p_pool_id, p_subscription_id, p_purchase_id, p_grant_id,
         p_quantity, 'active', v_activated_at, p_ended_at, v_set_id, p_product_id);

    FOR v_shape IN
        SELECT s.plan_ladder_id AS lid, s.rank AS new_rank
        FROM core.product_conferral_shapes s
        WHERE s.product_id = p_product_id AND s.plan_ladder_id IS NOT NULL
    LOOP
        v_new_junction_id := uuidv7();
        v_new_rank := v_shape.new_rank;

        -- The pre-supersession incumbent on this rung (captured above)
        -- supplies from_rank and the transition classification. Missing key
        -- and JSON null both read as NULL → 'initiate'.
        v_from_rank := (v_from_ranks ->> v_shape.lid::text)::INTEGER;

        v_type := CASE
            WHEN v_from_rank IS NULL THEN 'initiate'
            WHEN v_new_rank > v_from_rank THEN 'upgrade'
            WHEN v_new_rank < v_from_rank THEN 'downgrade'
            ELSE 'transfer'  -- equal rank ⇒ same product ⇒ cross-source handoff
        END;

        INSERT INTO core.pool_provision_ladders
            (provision_ladder_id, provision_id, plan_ladder_id, pool_id,
             status, activated_at, ended_at, product_id)
        VALUES
            (v_new_junction_id, v_new_provision_id, v_shape.lid, p_pool_id,
             'active', v_activated_at, p_ended_at, p_product_id);

        INSERT INTO core.pool_provision_transitions
            (pool_id, provision_id, plan_ladder_id, from_rank, to_rank,
             transition_type, actor_type, actor_id, reason, effective_at, provision_ladder_id)
        VALUES
            (p_pool_id, v_new_provision_id, v_shape.lid, v_from_rank, v_new_rank,
             v_type, p_actor_type, p_actor_id, p_reason, v_activated_at, v_new_junction_id);
    END LOOP;

    RETURN QUERY SELECT v_new_provision_id, 'created'::VARCHAR;
END;
$$;
-- +goose StatementEnd

-- +goose StatementBegin
CREATE FUNCTION core.end_conferral(
    p_subscription_id  UUID DEFAULT NULL,
    p_purchase_id      UUID DEFAULT NULL,
    p_grant_id         UUID DEFAULT NULL,
    p_provision_id     UUID DEFAULT NULL,
    p_actor_type       VARCHAR DEFAULT 'system',
    p_actor_id         UUID DEFAULT NULL,
    p_reason           TEXT DEFAULT NULL,
    p_effective_at     TIMESTAMPTZ DEFAULT NULL
) RETURNS SETOF UUID
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = core, pg_temp
AS $$
DECLARE
    v_selector_count INTEGER;
    v_target         RECORD;
    v_junction       RECORD;
    v_from_rank      INTEGER;
    v_effective_at   TIMESTAMPTZ;
BEGIN
    v_effective_at := COALESCE(p_effective_at, NOW());
    v_selector_count := (p_subscription_id IS NOT NULL)::int
                      + (p_purchase_id IS NOT NULL)::int
                      + (p_grant_id IS NOT NULL)::int
                      + (p_provision_id IS NOT NULL)::int;
    IF v_selector_count <> 1 THEN
        RAISE EXCEPTION 'end_conferral: exactly one of subscription/purchase/grant/provision must be supplied (got %)', v_selector_count;
    END IF;

    -- Explicit provision id on an already-ended provision is an error;
    -- a source arc that resolves to nothing live is SUCCESS (a revocation
    -- is recordable even when nothing is left to revoke).
    IF p_provision_id IS NOT NULL THEN
        PERFORM 1 FROM core.pool_provisions pp
        WHERE pp.provision_id = p_provision_id AND pp.status = 'ended';
        IF FOUND THEN
            RAISE EXCEPTION USING
                ERRCODE = 'CF004',
                MESSAGE = format('end_conferral: provision %s is already ended', p_provision_id);
        END IF;
    END IF;

    FOR v_target IN
        SELECT pp.provision_id AS id, pp.pool_id AS pool
        FROM core.pool_provisions pp
        WHERE pp.status <> 'ended'
          AND ((p_provision_id    IS NOT NULL AND pp.provision_id    = p_provision_id)
            OR (p_subscription_id IS NOT NULL AND pp.subscription_id = p_subscription_id)
            OR (p_purchase_id     IS NOT NULL AND pp.purchase_id     = p_purchase_id)
            OR (p_grant_id        IS NOT NULL AND pp.grant_id        = p_grant_id))
    LOOP
        PERFORM 1 FROM core.resource_pools rp WHERE rp.pool_id = v_target.pool FOR UPDATE;

        -- Resolution is by source, never by ladder occupancy: only rungs
        -- this provision itself holds are ended (finding #25's durable fix).
        FOR v_junction IN
            SELECT l.provision_ladder_id AS jid, l.plan_ladder_id AS lid, l.product_id AS pid
            FROM core.pool_provision_ladders l
            WHERE l.provision_id = v_target.id
              AND l.status <> 'ended'
        LOOP
            SELECT t.rank INTO v_from_rank
            FROM core.plan_ladder_tiers t
            WHERE t.plan_ladder_id = v_junction.lid AND t.product_id = v_junction.pid;

            INSERT INTO core.pool_provision_transitions
                (pool_id, provision_id, plan_ladder_id, from_rank, to_rank,
                 transition_type, actor_type, actor_id, reason, effective_at, provision_ladder_id)
            VALUES
                (v_target.pool, v_target.id, v_junction.lid, v_from_rank, NULL,
                 'end', p_actor_type, p_actor_id, p_reason, v_effective_at, v_junction.jid);
        END LOOP;

        UPDATE core.pool_provisions pp
        SET status = 'ended', ended_at = v_effective_at
        WHERE pp.provision_id = v_target.id;

        RETURN NEXT v_target.id;
    END LOOP;

    RETURN;
END;
$$;
-- +goose StatementEnd

-- +goose StatementBegin
CREATE FUNCTION core.sync_source_status(
    p_status           VARCHAR,
    p_subscription_id  UUID DEFAULT NULL,
    p_purchase_id      UUID DEFAULT NULL,
    p_grant_id         UUID DEFAULT NULL,
    p_effective_at     TIMESTAMPTZ DEFAULT NULL
) RETURNS SETOF UUID
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = core, pg_temp
AS $$
DECLARE
    v_source_count INTEGER;
BEGIN
    IF p_status NOT IN ('suspended', 'active') THEN
        RAISE EXCEPTION 'sync_source_status: status must be suspended or active (got %); ending is end_conferral''s job alone', p_status;
    END IF;
    v_source_count := (p_subscription_id IS NOT NULL)::int
                    + (p_purchase_id IS NOT NULL)::int
                    + (p_grant_id IS NOT NULL)::int;
    IF v_source_count <> 1 THEN
        RAISE EXCEPTION 'sync_source_status: exactly one of subscription/purchase/grant must be supplied (got %)', v_source_count;
    END IF;

    -- Never writes 'ended'; never resurrects — a provision superseded or
    -- ended while suspended stays ended (resume returns the empty set).
    RETURN QUERY
    UPDATE core.pool_provisions pp
    SET status = p_status,
        suspended_at = CASE WHEN p_status = 'suspended' THEN COALESCE(p_effective_at, NOW()) ELSE pp.suspended_at END
    WHERE pp.status <> 'ended'
      AND pp.status <> p_status
      AND ((p_subscription_id IS NOT NULL AND pp.subscription_id = p_subscription_id)
        OR (p_purchase_id     IS NOT NULL AND pp.purchase_id     = p_purchase_id)
        OR (p_grant_id        IS NOT NULL AND pp.grant_id        = p_grant_id))
    RETURNING pp.provision_id;
END;
$$;
-- +goose StatementEnd

-- +goose StatementBegin
CREATE FUNCTION core.align_conferral_shape(
    p_provision_id  UUID,
    p_actor_type    VARCHAR DEFAULT 'system',
    p_actor_id      UUID DEFAULT NULL,
    p_reason        TEXT DEFAULT NULL,
    p_effective_at  TIMESTAMPTZ DEFAULT NULL
) RETURNS TABLE (plan_ladder_id UUID, action VARCHAR)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = core, pg_temp
AS $$
DECLARE
    v_prov       RECORD;
    v_shape      RECORD;
    v_junction   RECORD;
    v_blocker    RECORD;
    v_rank       INTEGER;
    v_new_id     UUID;
    v_effective_at TIMESTAMPTZ;
BEGIN
    v_effective_at := COALESCE(p_effective_at, NOW());
    SELECT pp.provision_id AS id, pp.pool_id AS pool, pp.product_id AS pid,
           pp.status AS st, pp.ended_at AS ended
    INTO v_prov
    FROM core.pool_provisions pp
    WHERE pp.provision_id = p_provision_id;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'align_conferral_shape: provision % does not exist', p_provision_id;
    END IF;
    IF v_prov.st = 'ended' THEN
        RAISE EXCEPTION USING
            ERRCODE = 'CF004',
            MESSAGE = format('align_conferral_shape: provision %s is already ended', p_provision_id);
    END IF;

    PERFORM 1 FROM core.resource_pools rp WHERE rp.pool_id = v_prov.pool FOR UPDATE;

    -- Attach newly-required rungs (catalog gained a ladder). This is
    -- reconciliation of an existing conferral, not a new commercial event.
    FOR v_shape IN
        SELECT s.plan_ladder_id AS lid, s.rank AS r
        FROM core.product_conferral_shapes s
        WHERE s.product_id = v_prov.pid
          AND s.plan_ladder_id IS NOT NULL
          AND NOT EXISTS (
              SELECT 1 FROM core.pool_provision_ladders l
              WHERE l.provision_id = v_prov.id
                AND l.plan_ladder_id = s.plan_ladder_id
                AND l.status <> 'ended'
          )
    LOOP
        SELECT l.provision_id AS id, l.status AS st INTO v_blocker
        FROM core.pool_provision_ladders l
        WHERE l.pool_id = v_prov.pool
          AND l.plan_ladder_id = v_shape.lid
          AND l.status IN ('active', 'suspended')
          AND l.provision_id <> v_prov.id
        LIMIT 1;
        IF v_blocker.id IS NOT NULL THEN
            RAISE EXCEPTION USING
                ERRCODE = 'CF003',
                MESSAGE = format('align_conferral_shape: rung on ladder %s is held by provision %s (%s); alignment cannot supersede', v_shape.lid, v_blocker.id, v_blocker.st);
        END IF;

        v_new_id := uuidv7();
        INSERT INTO core.pool_provision_ladders
            (provision_ladder_id, provision_id, plan_ladder_id, pool_id,
             status, activated_at, ended_at, product_id)
        VALUES
            (v_new_id, v_prov.id, v_shape.lid, v_prov.pool,
             v_prov.st, v_effective_at, v_prov.ended, v_prov.pid);

        INSERT INTO core.pool_provision_transitions
            (pool_id, provision_id, plan_ladder_id, from_rank, to_rank,
             transition_type, actor_type, actor_id, reason, effective_at, provision_ladder_id)
        VALUES
            (v_prov.pool, v_prov.id, v_shape.lid, NULL, v_shape.r,
             'initiate', p_actor_type, p_actor_id, p_reason, v_effective_at, v_new_id);

        RETURN QUERY SELECT v_shape.lid, 'attached'::VARCHAR;
    END LOOP;

    -- End rungs the shape no longer includes (catalog lost a ladder).
    -- Individually ended: the monotonic-end guard keeps them ended.
    FOR v_junction IN
        SELECT l.provision_ladder_id AS jid, l.plan_ladder_id AS lid, l.product_id AS pid
        FROM core.pool_provision_ladders l
        WHERE l.provision_id = v_prov.id
          AND l.status <> 'ended'
          AND NOT EXISTS (
              SELECT 1 FROM core.product_conferral_shapes s
              WHERE s.product_id = v_prov.pid
                AND s.plan_ladder_id = l.plan_ladder_id
          )
    LOOP
        SELECT t.rank INTO v_rank
        FROM core.plan_ladder_tiers t
        WHERE t.plan_ladder_id = v_junction.lid AND t.product_id = v_junction.pid;

        INSERT INTO core.pool_provision_transitions
            (pool_id, provision_id, plan_ladder_id, from_rank, to_rank,
             transition_type, actor_type, actor_id, reason, effective_at, provision_ladder_id)
        VALUES
            (v_prov.pool, v_prov.id, v_junction.lid, v_rank, NULL,
             'end', p_actor_type, p_actor_id, p_reason, v_effective_at, v_junction.jid);

        UPDATE core.pool_provision_ladders l
        SET status = 'ended', ended_at = v_effective_at
        WHERE l.provision_ladder_id = v_junction.jid;

        RETURN QUERY SELECT v_junction.lid, 'ended'::VARCHAR;
    END LOOP;

    RETURN;
END;
$$;
-- +goose StatementEnd

-- +goose StatementBegin
CREATE FUNCTION core.update_conferral_bounds(
    p_provision_id  UUID,
    p_quantity      INTEGER DEFAULT NULL,
    p_ended_at      TIMESTAMPTZ DEFAULT NULL,
    p_set_ended_at  BOOLEAN DEFAULT FALSE,
    p_actor_type    VARCHAR DEFAULT 'system',
    p_actor_id      UUID DEFAULT NULL,
    p_reason        TEXT DEFAULT NULL
) RETURNS UUID
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = core, pg_temp
AS $$
DECLARE
    v_prov RECORD;
BEGIN
    SELECT pp.provision_id AS id, pp.status AS st, pp.grant_id AS gid, pp.pool_id AS pool
    INTO v_prov
    FROM core.pool_provisions pp
    WHERE pp.provision_id = p_provision_id;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'update_conferral_bounds: provision % does not exist', p_provision_id;
    END IF;
    IF v_prov.st = 'ended' THEN
        RAISE EXCEPTION USING
            ERRCODE = 'CF004',
            MESSAGE = format('update_conferral_bounds: provision %s is already ended', p_provision_id);
    END IF;
    -- Grant-sourced bounds change through extend-as-replace, preserving the
    -- decree ledger; this function is for commercial sources.
    IF v_prov.gid IS NOT NULL THEN
        RAISE EXCEPTION 'update_conferral_bounds: provision % is grant-sourced; use extend-as-replace', p_provision_id;
    END IF;
    IF p_quantity IS NOT NULL AND p_quantity < 1 THEN
        RAISE EXCEPTION 'update_conferral_bounds: quantity must be positive (got %)', p_quantity;
    END IF;

    PERFORM 1 FROM core.resource_pools rp WHERE rp.pool_id = v_prov.pool FOR UPDATE;

    -- A seat count or bound changing is not a position changing: no
    -- transition rows are written (quantity-blind audit trail, §7.2).
    UPDATE core.pool_provisions pp
    SET quantity = COALESCE(p_quantity, pp.quantity),
        ended_at = CASE WHEN p_set_ended_at THEN p_ended_at ELSE pp.ended_at END
    WHERE pp.provision_id = p_provision_id;

    RETURN p_provision_id;
END;
$$;
-- +goose StatementEnd

-- ---------------------------------------------------------------------------
-- The enclosure (Doc 41 §5.4). Functions run as core_owner; core_writer
-- keeps SELECT on the position tables but loses DML; EXECUTE is granted to
-- core_writer alone. The pinned search_path above is load-bearing
-- SECURITY DEFINER hardening, not decoration.
-- ---------------------------------------------------------------------------

ALTER FUNCTION core.confer(UUID, UUID, UUID, UUID, UUID, INTEGER, TIMESTAMPTZ, TIMESTAMPTZ, VARCHAR, UUID, TEXT) OWNER TO core_owner;
ALTER FUNCTION core.end_conferral(UUID, UUID, UUID, UUID, VARCHAR, UUID, TEXT, TIMESTAMPTZ) OWNER TO core_owner;
ALTER FUNCTION core.sync_source_status(VARCHAR, UUID, UUID, UUID, TIMESTAMPTZ) OWNER TO core_owner;
ALTER FUNCTION core.align_conferral_shape(UUID, VARCHAR, UUID, TEXT, TIMESTAMPTZ) OWNER TO core_owner;
ALTER FUNCTION core.update_conferral_bounds(UUID, INTEGER, TIMESTAMPTZ, BOOLEAN, VARCHAR, UUID, TEXT) OWNER TO core_owner;

REVOKE EXECUTE ON FUNCTION core.confer(UUID, UUID, UUID, UUID, UUID, INTEGER, TIMESTAMPTZ, TIMESTAMPTZ, VARCHAR, UUID, TEXT) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION core.end_conferral(UUID, UUID, UUID, UUID, VARCHAR, UUID, TEXT, TIMESTAMPTZ) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION core.sync_source_status(VARCHAR, UUID, UUID, UUID, TIMESTAMPTZ) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION core.align_conferral_shape(UUID, VARCHAR, UUID, TEXT, TIMESTAMPTZ) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION core.update_conferral_bounds(UUID, INTEGER, TIMESTAMPTZ, BOOLEAN, VARCHAR, UUID, TEXT) FROM PUBLIC;

GRANT EXECUTE ON FUNCTION core.confer(UUID, UUID, UUID, UUID, UUID, INTEGER, TIMESTAMPTZ, TIMESTAMPTZ, VARCHAR, UUID, TEXT) TO core_writer;
GRANT EXECUTE ON FUNCTION core.end_conferral(UUID, UUID, UUID, UUID, VARCHAR, UUID, TEXT, TIMESTAMPTZ) TO core_writer;
GRANT EXECUTE ON FUNCTION core.sync_source_status(VARCHAR, UUID, UUID, UUID, TIMESTAMPTZ) TO core_writer;
GRANT EXECUTE ON FUNCTION core.align_conferral_shape(UUID, VARCHAR, UUID, TEXT, TIMESTAMPTZ) TO core_writer;
GRANT EXECUTE ON FUNCTION core.update_conferral_bounds(UUID, INTEGER, TIMESTAMPTZ, BOOLEAN, VARCHAR, UUID, TEXT) TO core_writer;

REVOKE INSERT, UPDATE, DELETE ON core.pool_provisions FROM core_writer;
REVOKE INSERT, UPDATE, DELETE ON core.pool_provision_ladders FROM core_writer;
REVOKE INSERT, UPDATE, DELETE ON core.pool_provision_transitions FROM core_writer;

-- +goose Down

GRANT INSERT, UPDATE, DELETE ON core.pool_provisions TO core_writer;
GRANT INSERT, UPDATE, DELETE ON core.pool_provision_ladders TO core_writer;
GRANT INSERT, UPDATE, DELETE ON core.pool_provision_transitions TO core_writer;

DROP FUNCTION core.update_conferral_bounds(UUID, INTEGER, TIMESTAMPTZ, BOOLEAN, VARCHAR, UUID, TEXT);
DROP FUNCTION core.align_conferral_shape(UUID, VARCHAR, UUID, TEXT, TIMESTAMPTZ);
DROP FUNCTION core.sync_source_status(VARCHAR, UUID, UUID, UUID, TIMESTAMPTZ);
DROP FUNCTION core.end_conferral(UUID, UUID, UUID, UUID, VARCHAR, UUID, TEXT, TIMESTAMPTZ);
DROP FUNCTION core.confer(UUID, UUID, UUID, UUID, UUID, INTEGER, TIMESTAMPTZ, TIMESTAMPTZ, VARCHAR, UUID, TEXT);