supabase-brand-studio-crawl-task-rollout

-- Lead applies AFTER supabase-brand-studio-pull-tasks.sql. Never route existing
-- v1 jobs to v2. This function imports only an acknowledged parent reservation.
BEGIN;
CREATE OR REPLACE FUNCTION public.brand_studio_adopt_crawl_task(
  p_pull_id uuid,p_parent_token uuid,p_provider_task_id text,p_state jsonb,p_evidence_key uuid,p_payload_text text)
RETURNS SETOF public.brand_studio_pull_tasks LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $$
DECLARE j public.brand_studio_pulls; t public.brand_studio_pull_tasks;
  payload jsonb := p_payload_text::jsonb; purchase jsonb; unit jsonb; receipt jsonb;
BEGIN
  SELECT * INTO j FROM public.brand_studio_pulls WHERE id=p_pull_id FOR UPDATE;
  IF NOT FOUND OR j.status<>'running' OR p_parent_token IS NULL
    OR j.signals#>>'{queue,lease_token}' IS DISTINCT FROM p_parent_token::text
    OR j.signals#>>'{queue,lease_expires_at}' IS NULL
    OR (j.signals#>>'{queue,lease_expires_at}')::timestamptz<=clock_timestamp()
    OR j.signals#>>'{queue,job,executor_version}' IS DISTINCT FROM 'pull-stage/crawl-task-v2'
    OR j.signals#>>'{queue,reservation_pending}'='true' THEN RETURN; END IF;
  -- A lost adoption reply reconciles only this same immutable payload/key.
  SELECT * INTO t FROM public.brand_studio_pull_tasks WHERE pull_id=j.id AND task_key='crawl';
  IF FOUND THEN
    IF t.provider_task_id IS DISTINCT FROM p_provider_task_id OR NOT EXISTS (
      SELECT 1 FROM public.brand_studio_pull_task_evidence e WHERE e.task_id=t.id AND e.evidence_key=p_evidence_key AND e.payload_text=p_payload_text
    ) THEN RAISE EXCEPTION 'adoption_identity_conflict'; END IF;
    RETURN NEXT t; RETURN;
  END IF;
  IF payload->>'kind' IS DISTINCT FROM 'collector_resume'
    OR coalesce(payload->>'source_digest','') !~ '^[a-f0-9]{64}$'
    OR coalesce(payload->>'parent_plan_hash','') !~ '^[a-f0-9]{64}$'
    OR payload#>>'{resume,plan,domain}' IS DISTINCT FROM j.signals#>>'{queue,job,execution_plan,domain}'
    OR payload#>>'{resume,plan,plan_id}' IS DISTINCT FROM j.signals#>>'{queue,job,execution_plan,legs,site_evidence,plan_id}'
    OR (payload#>>'{resume,plan,max_cost_usd}')::numeric IS DISTINCT FROM (j.signals#>>'{queue,job,execution_plan,legs,site_evidence,max_cost_usd}')::numeric
    OR payload->>'parent_executor' IS DISTINCT FROM 'pull-stage/crawl-task-v2'
    OR payload->>'parent_plan_hash' IS DISTINCT FROM j.signals#>>'{queue,job,execution_plan,plan_hash}'
    OR payload->>'source_digest' IS DISTINCT FROM j.signals#>>'{queue,job,execution_plan,execution_contract,source_sha256}'
    OR payload->>'provider_task_id' IS DISTINCT FROM p_provider_task_id
    OR p_state->>'provider_task_id' IS DISTINCT FROM p_provider_task_id
    OR p_state->>'source_digest' IS DISTINCT FROM payload->>'source_digest'
    OR p_state->>'evidence_ref' IS DISTINCT FROM p_evidence_key::text
    OR jsonb_array_length(payload#>'{resume,units}')<>1
    OR coalesce((payload#>>'{resume,in_flight,purchase}')::boolean,false) THEN RAISE EXCEPTION 'invalid_crawl_adoption'; END IF;
  unit := payload#>'{resume,units,0}';
  IF unit->>'item_id' IS DISTINCT FROM 'crawl' OR unit->>'task_id' IS DISTINCT FROM p_provider_task_id
    OR NOT EXISTS (SELECT 1 FROM jsonb_array_elements(j.signals#>'{queue,job,execution_plan,legs,site_evidence,items}') i
      WHERE i->>'item_id'='crawl' AND (i->>'limit')::integer=(unit->>'limit')::integer)
    THEN RAISE EXCEPTION 'crawl_scope_mismatch'; END IF;
  IF (SELECT count(*) FROM jsonb_array_elements(payload#>'{resume,requests}') r WHERE r->>'purchase'='true' AND r->>'task_id'=p_provider_task_id)<>1
    THEN RAISE EXCEPTION 'accepted_purchase_receipt_required'; END IF;
  SELECT r INTO purchase FROM jsonb_array_elements(payload#>'{resume,requests}') r WHERE r->>'purchase'='true' AND r->>'task_id'=p_provider_task_id;
  receipt := j.signals#>'{queue,checkpoint,request_ledger,attempts}'->('site_evidence:'||(purchase->>'attempt_id'));
  IF NOT coalesce(unit->'request_ids' ? (purchase->>'attempt_id'),false) OR receipt IS NULL OR receipt->>'phase' NOT IN ('completed','uncertain')
    OR receipt->>'task_id' IS DISTINCT FROM p_provider_task_id OR receipt->>'purchase' IS DISTINCT FROM 'true'
    OR receipt->>'endpoint' IS DISTINCT FROM '/on_page/task_post' OR receipt->>'method' IS DISTINCT FROM 'POST'
    OR receipt->>'max_cost_usd' IS DISTINCT FROM purchase->>'max_cost_usd'
    OR receipt->>'actual_cost_usd' IS DISTINCT FROM purchase->>'actual_cost_usd'
    OR receipt->>'target' IS DISTINCT FROM payload#>>'{resume,plan,domain}'
    OR receipt->>'status' NOT IN ('passed','pending') THEN RAISE EXCEPTION 'parent_purchase_receipt_required'; END IF;
  INSERT INTO public.brand_studio_pull_tasks(pull_id,user_id,executor_version,task_key,provider_task_id,state)
    VALUES(j.id,j.user_id,'crawl-task-pilot/v1','crawl',p_provider_task_id,p_state) RETURNING * INTO t;
  INSERT INTO public.brand_studio_pull_task_evidence(task_id,evidence_key,payload_text,sha256)
    VALUES(t.id,p_evidence_key,p_payload_text,encode(sha256(convert_to(p_payload_text,'UTF8')),'hex'));
  RETURN NEXT t;
END $$;
REVOKE ALL ON FUNCTION public.brand_studio_adopt_crawl_task(uuid,uuid,text,jsonb,uuid,text) FROM PUBLIC,anon,authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_adopt_crawl_task(uuid,uuid,text,jsonb,uuid,text) TO service_role;
NOTIFY pgrst,'reload schema';
COMMIT;

-- Readbacks (lead runs after both migrations):
-- SELECT table_name FROM information_schema.tables WHERE table_schema='public' AND table_name LIKE 'brand_studio_pull_task%';
-- SELECT routine_name FROM information_schema.routines WHERE routine_schema='public' AND routine_name LIKE 'brand_studio_%crawl_task%';
-- SELECT has_function_privilege('service_role','public.brand_studio_adopt_crawl_task(uuid,uuid,text,jsonb,uuid,text)','EXECUTE'),
--   has_function_privilege('anon','public.brand_studio_adopt_crawl_task(uuid,uuid,text,jsonb,uuid,text)','EXECUTE'),
--   has_function_privilege('authenticated','public.brand_studio_adopt_crawl_task(uuid,uuid,text,jsonb,uuid,text)','EXECUTE'),
--   has_table_privilege('service_role','public.brand_studio_pull_tasks','UPDATE'); -- t,f,f,f
-- SELECT relname,relrowsecurity FROM pg_class WHERE relname LIKE 'brand_studio_pull_task%';