supabase-brand-studio-pull-tasks

-- Offline crawl pilot only. Lead applies after review; no automatic adoption.
-- PostgreSQL 17 row locks/SKIP LOCKED and function privileges, checked 2026-10-23:
-- https://www.postgresql.org/docs/17/sql-select.html
-- https://www.postgresql.org/docs/17/sql-createfunction.html
BEGIN;

CREATE TABLE IF NOT EXISTS public.brand_studio_pull_tasks (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  pull_id uuid NOT NULL REFERENCES public.brand_studio_pulls(id) ON DELETE CASCADE,
  user_id uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
  executor_version text NOT NULL CHECK (executor_version = 'crawl-task-pilot/v1'),
  task_key text NOT NULL CHECK (task_key = 'crawl'),
  provider_task_id text NOT NULL CHECK (length(provider_task_id) BETWEEN 1 AND 200),
  status text NOT NULL DEFAULT 'ready' CHECK (status IN ('ready','waiting','complete','failed')),
  next_run_at timestamptz,
  lease_token uuid,
  lease_until timestamptz,
  revision bigint NOT NULL DEFAULT 0,
  state jsonb NOT NULL DEFAULT '{}'::jsonb CHECK (jsonb_typeof(state) = 'object' AND octet_length(state::text)<=65536),
  created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
  updated_at timestamptz NOT NULL DEFAULT clock_timestamp(),
  UNIQUE(pull_id, task_key)
);
CREATE INDEX IF NOT EXISTS brand_studio_pull_tasks_due_idx ON public.brand_studio_pull_tasks(next_run_at,created_at,id) WHERE status IN ('ready','waiting');

CREATE TABLE IF NOT EXISTS public.brand_studio_pull_task_evidence (
  task_id uuid NOT NULL REFERENCES public.brand_studio_pull_tasks(id) ON DELETE CASCADE,
  evidence_key uuid NOT NULL,
  payload_text text NOT NULL CHECK (octet_length(payload_text) <= 12582912 AND jsonb_typeof(payload_text::jsonb) = 'object'),
  sha256 text NOT NULL,
  created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
  PRIMARY KEY(task_id,evidence_key),
  CHECK (sha256 = encode(sha256(convert_to(payload_text,'UTF8')),'hex'))
);
CREATE TABLE IF NOT EXISTS public.brand_studio_pull_task_transitions (
  task_id uuid NOT NULL REFERENCES public.brand_studio_pull_tasks(id) ON DELETE CASCADE,
  operation_id uuid NOT NULL,
  operation_digest text NOT NULL,
  revision bigint NOT NULL,
  PRIMARY KEY(task_id,operation_id)
);

ALTER TABLE public.brand_studio_pull_tasks ENABLE ROW LEVEL SECURITY;
ALTER TABLE public.brand_studio_pull_task_evidence ENABLE ROW LEVEL SECURITY;
ALTER TABLE public.brand_studio_pull_task_transitions ENABLE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS pull_tasks_owner_read ON public.brand_studio_pull_tasks;
CREATE POLICY pull_tasks_owner_read ON public.brand_studio_pull_tasks FOR SELECT TO authenticated USING (user_id = auth.uid());
DROP POLICY IF EXISTS pull_task_evidence_owner_read ON public.brand_studio_pull_task_evidence;
CREATE POLICY pull_task_evidence_owner_read ON public.brand_studio_pull_task_evidence FOR SELECT TO authenticated USING (EXISTS (SELECT 1 FROM public.brand_studio_pull_tasks t WHERE t.id=task_id AND t.user_id=auth.uid()));
REVOKE ALL ON public.brand_studio_pull_tasks, public.brand_studio_pull_task_evidence, public.brand_studio_pull_task_transitions FROM anon,authenticated;
GRANT SELECT ON public.brand_studio_pull_tasks, public.brand_studio_pull_task_evidence TO authenticated;
-- No direct mutation grant, even for service_role: all transitions use the RPC.
REVOKE ALL ON public.brand_studio_pull_tasks, public.brand_studio_pull_task_evidence, public.brand_studio_pull_task_transitions FROM service_role;
GRANT SELECT ON public.brand_studio_pull_tasks, public.brand_studio_pull_task_evidence TO service_role;

-- Import an ALREADY OWNED identity. Only an explicitly routed, unleased pull
-- can be adopted. Caller must supply the one receipt-bound task ID, never POST.
CREATE OR REPLACE FUNCTION public.brand_studio_seed_crawl_task(p_pull_id 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;
BEGIN
  SELECT * INTO j FROM public.brand_studio_pulls WHERE id=p_pull_id FOR UPDATE;
  IF NOT FOUND OR j.signals#>>'{queue,job,executor_version}' IS DISTINCT FROM 'crawl-task-pilot/v1'
    OR j.signals#>>'{queue,lease_token}' IS NOT NULL OR j.status <> 'running' THEN
    RAISE EXCEPTION 'pilot_route_required';
  END IF;
  IF p_state->>'provider_task_id' IS DISTINCT FROM p_provider_task_id THEN RAISE EXCEPTION 'task_identity_mismatch'; 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) ON CONFLICT(pull_id,task_key) DO NOTHING;
  SELECT * INTO t FROM public.brand_studio_pull_tasks WHERE pull_id=j.id AND task_key='crawl';
  IF t.provider_task_id IS DISTINCT FROM p_provider_task_id THEN RAISE EXCEPTION 'task_identity_mismatch'; END IF;
  IF t.state IS DISTINCT FROM p_state OR p_state->>'evidence_ref' IS DISTINCT FROM p_evidence_key::text THEN RAISE EXCEPTION 'seed_identity_conflict'; END IF;
  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')) ON CONFLICT DO NOTHING;
  IF 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 'seed_identity_conflict'; END IF;
  RETURN NEXT t;
END $$;

CREATE OR REPLACE FUNCTION public.brand_studio_claim_crawl_task(p_task_id uuid DEFAULT NULL,p_lease_seconds integer DEFAULT 120)
RETURNS SETOF public.brand_studio_pull_tasks LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $$
DECLARE t public.brand_studio_pull_tasks; at_time timestamptz := clock_timestamp();
BEGIN
  IF p_lease_seconds IS NULL OR p_lease_seconds NOT BETWEEN 1 AND 600 THEN RAISE EXCEPTION 'invalid_lease'; END IF;
  SELECT q.* INTO t FROM public.brand_studio_pull_tasks q JOIN public.brand_studio_pulls j ON j.id=q.pull_id
    WHERE (p_task_id IS NULL OR q.id=p_task_id) AND q.executor_version='crawl-task-pilot/v1'
    AND j.signals#>>'{queue,job,executor_version}' IN ('crawl-task-pilot/v1','pull-stage/crawl-task-v2') AND j.status='running'
    AND q.status IN ('ready','waiting') AND (q.next_run_at IS NULL OR q.next_run_at<=at_time)
    AND (q.lease_until IS NULL OR q.lease_until<=at_time)
    ORDER BY q.next_run_at NULLS FIRST,q.created_at,q.id FOR UPDATE OF q SKIP LOCKED LIMIT 1;
  IF NOT FOUND THEN RETURN; END IF;
  UPDATE public.brand_studio_pull_tasks SET lease_token=gen_random_uuid(),lease_until=at_time+make_interval(secs=>p_lease_seconds),updated_at=at_time
    WHERE id=t.id RETURNING * INTO t;
  RETURN NEXT t;
END $$;

-- Evidence insertion, state pointer and operation receipt commit atomically.
-- Retry the SAME operation id after an unknown response; it cannot apply twice.
CREATE OR REPLACE FUNCTION public.brand_studio_transition_crawl_task(p_task_id uuid,p_token uuid,p_revision bigint,p_operation_id uuid,
  p_state jsonb,p_status text,p_next_run_at timestamptz DEFAULT NULL,p_evidence_key uuid DEFAULT NULL,p_payload_text text DEFAULT NULL,p_release boolean DEFAULT true)
RETURNS SETOF public.brand_studio_pull_tasks LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $$
DECLARE t public.brand_studio_pull_tasks; prior public.brand_studio_pull_task_transitions; digest text;
BEGIN
  digest := encode(sha256(convert_to(jsonb_build_array(p_revision,p_state,p_status,p_next_run_at,p_evidence_key,p_payload_text,p_release)::text,'UTF8')),'hex');
  SELECT * INTO t FROM public.brand_studio_pull_tasks WHERE id=p_task_id FOR UPDATE;
  IF NOT FOUND THEN RETURN; END IF;
  SELECT * INTO prior FROM public.brand_studio_pull_task_transitions WHERE task_id=t.id AND operation_id=p_operation_id;
  IF FOUND THEN
    IF prior.operation_digest<>digest THEN RAISE EXCEPTION 'operation_identity_conflict'; END IF;
    RETURN NEXT t; RETURN;
  END IF;
  IF p_token IS NULL OR p_revision IS NULL OR t.lease_until IS NULL OR t.lease_token IS DISTINCT FROM p_token OR t.revision<>p_revision OR t.lease_until<=clock_timestamp()
    OR NOT EXISTS (SELECT 1 FROM public.brand_studio_pulls j WHERE j.id=t.pull_id AND j.status='running'
      AND j.signals#>>'{queue,job,executor_version}' IN ('crawl-task-pilot/v1','pull-stage/crawl-task-v2')) THEN RETURN; END IF;
  IF p_state->>'provider_task_id' IS DISTINCT FROM t.provider_task_id OR p_status NOT IN ('ready','waiting','complete','failed')
    OR (p_status='waiting' AND p_next_run_at IS NULL) OR (p_status<>'waiting' AND p_next_run_at IS NOT NULL)
    OR ((p_evidence_key IS NULL) <> (p_payload_text IS NULL)) THEN RAISE EXCEPTION 'invalid_task_transition'; END IF;
  IF p_evidence_key IS NOT NULL THEN
    IF p_state->>'evidence_ref' IS DISTINCT FROM p_evidence_key::text THEN RAISE EXCEPTION 'evidence_reference_mismatch'; END IF;
    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')) ON CONFLICT DO NOTHING;
    IF 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 'evidence_identity_conflict';
    END IF;
  END IF;
  IF NOT EXISTS (SELECT 1 FROM public.brand_studio_pull_task_evidence e WHERE e.task_id=t.id AND e.evidence_key::text=p_state->>'evidence_ref') THEN RAISE EXCEPTION 'evidence_reference_missing'; END IF;
  UPDATE public.brand_studio_pull_tasks SET state=p_state,status=p_status,next_run_at=p_next_run_at,revision=revision+1,
    lease_token=CASE WHEN p_release THEN NULL ELSE lease_token END,
    lease_until=CASE WHEN p_release THEN NULL ELSE lease_until END,updated_at=clock_timestamp() WHERE id=t.id RETURNING * INTO t;
  INSERT INTO public.brand_studio_pull_task_transitions VALUES(t.id,p_operation_id,digest,t.revision);
  RETURN NEXT t;
END $$;

REVOKE ALL ON FUNCTION public.brand_studio_seed_crawl_task(uuid,text,jsonb,uuid,text), public.brand_studio_claim_crawl_task(uuid,integer), public.brand_studio_transition_crawl_task(uuid,uuid,bigint,uuid,jsonb,text,timestamptz,uuid,text,boolean) FROM PUBLIC,anon,authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_seed_crawl_task(uuid,text,jsonb,uuid,text), public.brand_studio_claim_crawl_task(uuid,integer), public.brand_studio_transition_crawl_task(uuid,uuid,bigint,uuid,jsonb,text,timestamptz,uuid,text,boolean) TO service_role;
NOTIFY pgrst,'reload schema';
COMMIT;