supabase-brand-studio-alpha-admission

-- Atomic self-serve admission. Apply after pulls, api-keys and service-writes.
-- No provider calls, existing-row updates, credit changes or scheduler enablement.
BEGIN;
SET LOCAL lock_timeout = '5s';
SET LOCAL statement_timeout = '30s';

ALTER TABLE public.brand_studio_pulls
  ADD COLUMN IF NOT EXISTS alpha_reserved_max_usd numeric;
-- Reapplication upgrades the original positive-only constraint without
-- rewriting any reservation. Explicitly priced free pulls still consume an
-- account start and use the same lock/idempotency/credit guards.
ALTER TABLE public.brand_studio_pulls
  DROP CONSTRAINT IF EXISTS brand_studio_pulls_alpha_reserved_max_usd_check;
ALTER TABLE public.brand_studio_pulls
  ADD CONSTRAINT brand_studio_pulls_alpha_reserved_max_usd_check
  CHECK (alpha_reserved_max_usd IS NULL OR
    (alpha_reserved_max_usd >= 0 AND alpha_reserved_max_usd::text NOT IN ('NaN','Infinity','-Infinity')));

CREATE INDEX IF NOT EXISTS brand_studio_pulls_alpha_created_idx
  ON public.brand_studio_pulls (created_at)
  WHERE alpha_reserved_max_usd IS NOT NULL OR signals->>'idempotency_key' LIKE 'alpha-self-serve:%';

-- The persisted alpha key/amount, rather than caller intent, identifies this
-- lane. A forgotten legacy INSERT fails instead of bypassing the shared lock.
CREATE OR REPLACE FUNCTION public.brand_studio_guard_alpha_admission_v1()
RETURNS trigger LANGUAGE plpgsql SECURITY INVOKER SET search_path = '' AS $$
BEGIN
  IF TG_OP = 'UPDATE' THEN
    IF (NEW.alpha_reserved_max_usd IS NOT NULL OR NEW.signals->>'idempotency_key' LIKE 'alpha-self-serve:%')
      AND NOT (OLD.alpha_reserved_max_usd IS NOT NULL OR coalesce(OLD.signals->>'idempotency_key','') LIKE 'alpha-self-serve:%') THEN
      RAISE EXCEPTION 'alpha reservation requires atomic admission' USING ERRCODE = '42501';
    END IF;
    IF NEW.alpha_reserved_max_usd IS DISTINCT FROM OLD.alpha_reserved_max_usd THEN
      RAISE EXCEPTION 'alpha reservation amount is immutable' USING ERRCODE = '42501';
    END IF;
    IF OLD.alpha_reserved_max_usd IS NOT NULL OR OLD.signals->>'idempotency_key' LIKE 'alpha-self-serve:%' THEN
      IF NEW.signals->>'idempotency_key' IS DISTINCT FROM OLD.signals->>'idempotency_key'
        OR NEW.user_id IS DISTINCT FROM OLD.user_id OR NEW.created_at IS DISTINCT FROM OLD.created_at
        OR (OLD.signals#>>'{queue,reservation_pending}' = 'false'
          AND NEW.signals#>>'{queue,reservation_pending}' IS DISTINCT FROM 'false') THEN
        RAISE EXCEPTION 'alpha reservation identity and armed state are immutable' USING ERRCODE = '42501';
      END IF;
    END IF;
  ELSIF NEW.alpha_reserved_max_usd IS NOT NULL OR NEW.signals->>'idempotency_key' LIKE 'alpha-self-serve:%' THEN
    IF pg_catalog.current_setting('brand_studio.alpha_reservation', true)
      IS DISTINCT FROM NEW.user_id::text || ':' || (NEW.signals->>'idempotency_key')
      OR NEW.alpha_reserved_max_usd IS NULL THEN
      RAISE EXCEPTION 'alpha reservation requires atomic admission' USING ERRCODE = '42501';
    END IF;
  END IF;
  RETURN NEW;
END;
$$;
REVOKE ALL ON FUNCTION public.brand_studio_guard_alpha_admission_v1() FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_guard_alpha_admission_v1() TO service_role;
DROP TRIGGER IF EXISTS brand_studio_guard_alpha_admission_v1 ON public.brand_studio_pulls;
CREATE TRIGGER brand_studio_guard_alpha_admission_v1 BEFORE INSERT OR UPDATE
  ON public.brand_studio_pulls FOR EACH ROW EXECUTE FUNCTION public.brand_studio_guard_alpha_admission_v1();

CREATE OR REPLACE FUNCTION public.brand_studio_reserve_alpha_pull_v1(
  p_pull jsonb, p_per_pull_usd numeric, p_account_limit bigint, p_daily_usd numeric
) RETURNS jsonb LANGUAGE plpgsql VOLATILE SECURITY INVOKER
SET search_path = '' SET lock_timeout = '5s' AS $$
DECLARE
  owner_id uuid;
  request_key text := p_pull#>>'{signals,idempotency_key}';
  approved_max numeric;
  plan jsonb := p_pull#>'{signals,queue,job,execution_plan}';
  existing public.brand_studio_pulls%ROWTYPE;
  reserved public.brand_studio_pulls%ROWTYPE;
  account_starts bigint;
  committed numeric;
  admitted_at timestamptz;
  previous_marker text;
BEGIN
  -- VOLATILE queries obtain fresh snapshots after the transaction lock. A
  -- caller configured for repeatable-read must not sum an earlier snapshot.
  IF pg_catalog.current_setting('transaction_isolation') <> 'read committed' THEN
    RAISE EXCEPTION 'alpha admission requires read committed' USING ERRCODE = '25000';
  END IF;
  -- Only the service role can call this RPC. A null lifetime count requires
  -- the server's explicit account entitlement marker; all dollar caps remain.
  IF p_per_pull_usd IS NULL OR p_daily_usd IS NULL
    OR (p_account_limit IS NULL AND p_pull#>'{signals,account_unlimited}' IS DISTINCT FROM 'true'::jsonb)
    OR p_per_pull_usd::text IN ('NaN','Infinity','-Infinity')
    OR p_daily_usd::text IN ('NaN','Infinity','-Infinity')
    OR p_per_pull_usd <= 0 OR p_daily_usd < p_per_pull_usd
    OR p_account_limit <= 0 OR p_account_limit > 9007199254740991 THEN
    RETURN pg_catalog.jsonb_build_object('ok',false,'reason','alpha_allowance_not_configured');
  END IF;
  IF pg_catalog.jsonb_typeof(p_pull) IS DISTINCT FROM 'object'
    OR p_pull->>'status' IS DISTINCT FROM 'running' OR p_pull->>'route' IS DISTINCT FROM 'hosted'
    OR p_pull#>'{signals,queue,reservation_pending}' IS DISTINCT FROM 'true'::jsonb
    OR pg_catalog.jsonb_typeof(plan->'max_cost_usd') IS DISTINCT FROM 'number'
    OR (plan->>'plan_hash') IS NULL OR (plan->>'plan_hash') !~ '^[0-9a-f]{64}$'
    OR p_pull#>'{signals,execution_plan}' IS DISTINCT FROM plan
    OR request_key IS NULL OR request_key !~ '^alpha-self-serve:[0-9a-f]{64}:[0-9]{4}-[0-9]{2}-[0-9]{2}(:[A-Za-z0-9_.:-]{1,60})?$'
    OR pg_catalog.split_part(request_key,':',2) IS DISTINCT FROM plan->>'plan_hash'
    OR coalesce(p_pull->>'run_id','') = '' THEN
    RETURN pg_catalog.jsonb_build_object('ok',false,'reason','alpha_reservation_invalid');
  END IF;
  owner_id := (p_pull->>'user_id')::uuid;
  approved_max := (plan->>'max_cost_usd')::numeric;
  IF owner_id IS NULL OR approved_max < 0 THEN
    RETURN pg_catalog.jsonb_build_object('ok',false,'reason','alpha_plan_unpriced');
  END IF;
  IF approved_max > p_per_pull_usd THEN
    RETURN pg_catalog.jsonb_build_object('ok',false,'reason','alpha_report_over_limit');
  END IF;

  -- One lock for ALL accounts. The maximum wait is five seconds, with no
  -- retry inside the RPC or application after an uncertain mutation.
  PERFORM pg_catalog.pg_advisory_xact_lock(1112757313,1);
  admitted_at := pg_catalog.clock_timestamp();
  SELECT * INTO existing FROM public.brand_studio_pulls
    WHERE user_id = owner_id AND signals->>'idempotency_key' = request_key;
  IF FOUND THEN
    RETURN pg_catalog.jsonb_build_object('ok',true,'pull',pg_catalog.to_jsonb(existing),'replayed',true);
  END IF;

  -- Account starts exclude only terminal, provably never-armed reservations.
  -- Dollar reservations remain committed for 24h even after a confirmed void.
  SELECT count(*) INTO account_starts FROM public.brand_studio_pulls
    WHERE user_id = owner_id
      AND (alpha_reserved_max_usd IS NOT NULL OR signals->>'idempotency_key' LIKE 'alpha-self-serve:%')
      AND NOT (status = 'failed' AND coalesce(signals#>>'{queue,reservation_pending}','') = 'true');
  IF p_account_limit IS NOT NULL AND account_starts >= p_account_limit THEN
    RETURN pg_catalog.jsonb_build_object('ok',false,'reason','alpha_account_allowance_used');
  END IF;

  -- Bounded numeric syntax makes the cast safe, including arbitrary legacy
  -- JSON strings. Zero, negative, nonnumeric, special and absurd values count
  -- at the full cap. SUM runs in SQL and has no REST pagination ceiling.
  SELECT coalesce(sum(coalesce(alpha_reserved_max_usd,
    CASE WHEN pg_catalog.length(legacy_max) <= 64
      AND legacy_max ~ '^[0-9]{1,12}(\.[0-9]{1,30})?([eE][+-]?[0-9]{1,2})?$'
      THEN CASE WHEN legacy_max::numeric > 0 AND legacy_max::numeric <= 1000000000
        THEN legacy_max::numeric ELSE p_per_pull_usd END
      ELSE p_per_pull_usd END)),0) INTO committed
    FROM (SELECT alpha_reserved_max_usd, signals#>>'{queue,job,execution_plan,max_cost_usd}' AS legacy_max
      FROM public.brand_studio_pulls
      WHERE (alpha_reserved_max_usd IS NOT NULL OR signals->>'idempotency_key' LIKE 'alpha-self-serve:%')
        AND created_at >= admitted_at - interval '24 hours') recent;
  IF committed + approved_max > p_daily_usd THEN
    RETURN pg_catalog.jsonb_build_object('ok',false,'reason','alpha_daily_budget_reached');
  END IF;

  previous_marker := pg_catalog.current_setting('brand_studio.alpha_reservation',true);
  PERFORM pg_catalog.set_config('brand_studio.alpha_reservation',owner_id::text || ':' || request_key,true);
  INSERT INTO public.brand_studio_pulls
    (user_id,brand_id,run_id,source_url,domain,title,route,status,pull_profile_id,signals,created_at,alpha_reserved_max_usd)
    VALUES (owner_id,(p_pull->>'brand_id')::uuid,p_pull->>'run_id',p_pull->>'source_url',p_pull->>'domain',
      p_pull->>'title','hosted','running',coalesce(p_pull->>'pull_profile_id','standard'),p_pull->'signals',admitted_at,approved_max)
    ON CONFLICT (user_id,(signals->>'idempotency_key')) WHERE signals->>'idempotency_key' IS NOT NULL
      DO NOTHING RETURNING * INTO reserved;
  PERFORM pg_catalog.set_config('brand_studio.alpha_reservation',coalesce(previous_marker,''),true);
  IF reserved.id IS NULL THEN
    SELECT * INTO existing FROM public.brand_studio_pulls
      WHERE user_id = owner_id AND signals->>'idempotency_key' = request_key;
    IF NOT FOUND THEN RAISE EXCEPTION 'alpha reservation was not persisted'; END IF;
    RETURN pg_catalog.jsonb_build_object('ok',true,'pull',pg_catalog.to_jsonb(existing),'replayed',true);
  END IF;
  RETURN pg_catalog.jsonb_build_object('ok',true,'pull',pg_catalog.to_jsonb(reserved),'replayed',false);
END;
$$;
REVOKE ALL ON FUNCTION public.brand_studio_reserve_alpha_pull_v1(jsonb,numeric,bigint,numeric) FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_reserve_alpha_pull_v1(jsonb,numeric,bigint,numeric) TO service_role;
NOTIFY pgrst, 'reload schema';
COMMIT;