supabase-brand-studio-evidence-chunks

-- Optional NEW RUN ONLY evidence_storage='chunks_v1'. Apply after pulls schema.
-- No backfill, reparenting, provider calls, or checkpoint/cursor writes here.
-- A caller must await every {sha256} receipt before its final fenced pull PATCH
-- atomically publishes the receipt/cursor and references. Orphan chunks after a
-- crash DO NOT reconcile an uncertain response or authorize repurchase. Keep
-- unknown/no-repurchase guards; read-only recovery is separate from collection.
-- Local disposable PostgreSQL proof is not a deployment or pipeline proof.
BEGIN;
CREATE EXTENSION IF NOT EXISTS pgcrypto;
CREATE UNIQUE INDEX IF NOT EXISTS brand_studio_pulls_id_user_idx
  ON public.brand_studio_pulls(id, user_id);

-- Resolve pgcrypto's installed schema once. Supabase can install it in
-- extensions rather than public; generated functions use fully qualified names.
DO $migration$
DECLARE crypto_schema text;
BEGIN
  SELECT n.nspname INTO STRICT crypto_schema FROM pg_catalog.pg_extension e
    JOIN pg_catalog.pg_namespace n ON n.oid = e.extnamespace WHERE e.extname = 'pgcrypto';
  EXECUTE format($ddl$
    CREATE TABLE IF NOT EXISTS public.brand_studio_evidence_chunks (
      pull_id uuid NOT NULL,
      user_id uuid NOT NULL,
      sha256 text NOT NULL CHECK (sha256 ~ '^[a-f0-9]{64}$'),
      payload_base64 text NOT NULL,
      byte_length integer GENERATED ALWAYS AS (octet_length(decode(payload_base64, 'base64'))) STORED,
      created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
      PRIMARY KEY (pull_id, user_id, sha256),
      CONSTRAINT brand_studio_evidence_chunks_owner_fk FOREIGN KEY (pull_id, user_id)
        REFERENCES public.brand_studio_pulls(id, user_id) ON DELETE CASCADE,
      CONSTRAINT brand_studio_evidence_chunks_size CHECK (byte_length BETWEEN 1 AND 262144),
      CONSTRAINT brand_studio_evidence_chunks_base64 CHECK (
        length(payload_base64) <= 349528 AND
        payload_base64 = replace(encode(decode(payload_base64, 'base64'), 'base64'), E'\n', '')),
      CONSTRAINT brand_studio_evidence_chunks_hash CHECK (
        sha256 = encode(%I.digest(decode(payload_base64, 'base64'), 'sha256'), 'hex'))
    )
  $ddl$, crypto_schema);

  EXECUTE format($function$
    CREATE OR REPLACE FUNCTION public.brand_studio_store_evidence_chunk_v1(
      p_pull_id uuid, p_user_id uuid, p_lease_token text, p_sha256 text, p_payload_base64 text
    ) RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path = '' AS $body$
    DECLARE
      owned_pull public.brand_studio_pulls%%ROWTYPE;
      payload bytea;
      expiry timestamptz;
      existing_payload text;
    BEGIN
      IF p_pull_id IS NULL OR p_user_id IS NULL OR p_lease_token IS NULL OR btrim(p_lease_token) = '' THEN
        RAISE EXCEPTION 'evidence_chunk_invalid_scope_or_lease' USING ERRCODE = '22023';
      END IF;
      -- Same row locked by queue checkpoint updates: lease validation and insert
      -- cannot race a reclaim or terminal transition. Check time AFTER waiting.
      SELECT * INTO owned_pull FROM public.brand_studio_pulls
        WHERE id = p_pull_id AND user_id = p_user_id FOR UPDATE;
      IF NOT FOUND OR owned_pull.status IS DISTINCT FROM 'running'
        OR owned_pull.signals #>> '{queue,lease_token}' IS DISTINCT FROM p_lease_token THEN
        RAISE EXCEPTION 'evidence_chunk_stale_lease_or_owner' USING ERRCODE = '42501';
      END IF;
      BEGIN
        expiry := (owned_pull.signals #>> '{queue,lease_expires_at}')::timestamptz;
      EXCEPTION WHEN invalid_datetime_format OR datetime_field_overflow THEN
        RAISE EXCEPTION 'evidence_chunk_invalid_lease_expiry' USING ERRCODE = '42501';
      END;
      IF expiry IS NULL OR NOT isfinite(expiry) OR expiry <= clock_timestamp() THEN
        RAISE EXCEPTION 'evidence_chunk_expired_lease' USING ERRCODE = '42501';
      END IF;
      IF p_sha256 IS NULL OR p_sha256 !~ '^[a-f0-9]{64}$'
        OR p_payload_base64 IS NULL OR length(p_payload_base64) NOT BETWEEN 4 AND 349528
        OR length(p_payload_base64) %% 4 <> 0 OR p_payload_base64 !~ '^[A-Za-z0-9+/]*={0,2}$' THEN
        RAISE EXCEPTION 'evidence_chunk_invalid_payload' USING ERRCODE = '22023';
      END IF;
      payload := decode(p_payload_base64, 'base64');
      IF octet_length(payload) NOT BETWEEN 1 AND 262144
        OR replace(encode(payload, 'base64'), E'\n', '') <> p_payload_base64
        OR encode(%I.digest(payload, 'sha256'), 'hex') <> p_sha256 THEN
        RAISE EXCEPTION 'evidence_chunk_size_or_hash_mismatch' USING ERRCODE = '22023';
      END IF;
      -- No UPDATE on conflict: even metadata is immutable. Retried writes must
      -- match exact canonical bytes and still hold a current active lease.
      INSERT INTO public.brand_studio_evidence_chunks(pull_id, user_id, sha256, payload_base64)
        VALUES(p_pull_id, p_user_id, p_sha256, p_payload_base64)
        ON CONFLICT (pull_id, user_id, sha256) DO NOTHING;
      SELECT payload_base64 INTO STRICT existing_payload FROM public.brand_studio_evidence_chunks
        WHERE pull_id = p_pull_id AND user_id = p_user_id AND sha256 = p_sha256;
      IF existing_payload IS DISTINCT FROM p_payload_base64 THEN
        RAISE EXCEPTION 'evidence_chunk_immutable_collision' USING ERRCODE = '23505';
      END IF;
      RETURN jsonb_build_object('sha256', p_sha256);
    END;
    $body$;
  $function$, crypto_schema);
END;
$migration$;

CREATE OR REPLACE FUNCTION public.brand_studio_evidence_chunks_no_update()
RETURNS trigger LANGUAGE plpgsql SET search_path = '' AS $$
BEGIN
  RAISE EXCEPTION 'evidence_chunk_immutable' USING ERRCODE = '55000';
END;
$$;
REVOKE ALL ON FUNCTION public.brand_studio_evidence_chunks_no_update() FROM PUBLIC, anon, authenticated, service_role;
DROP TRIGGER IF EXISTS brand_studio_evidence_chunks_immutable ON public.brand_studio_evidence_chunks;
CREATE TRIGGER brand_studio_evidence_chunks_immutable BEFORE UPDATE ON public.brand_studio_evidence_chunks
  FOR EACH ROW EXECUTE FUNCTION public.brand_studio_evidence_chunks_no_update();

ALTER TABLE public.brand_studio_evidence_chunks ENABLE ROW LEVEL SECURITY;
REVOKE ALL ON public.brand_studio_evidence_chunks FROM PUBLIC, anon, authenticated, service_role;
GRANT SELECT ON public.brand_studio_evidence_chunks TO authenticated, service_role;
DROP POLICY IF EXISTS evidence_chunks_owner_read ON public.brand_studio_evidence_chunks;
CREATE POLICY evidence_chunks_owner_read ON public.brand_studio_evidence_chunks
  FOR SELECT TO authenticated USING (user_id = (SELECT auth.uid()));
-- Restrictive policies also guard unexpected inherited client grants.
DROP POLICY IF EXISTS evidence_chunks_owner_bound ON public.brand_studio_evidence_chunks;
CREATE POLICY evidence_chunks_owner_bound ON public.brand_studio_evidence_chunks AS RESTRICTIVE
  FOR SELECT TO authenticated USING (user_id = (SELECT auth.uid()));
DROP POLICY IF EXISTS evidence_chunks_no_client_insert ON public.brand_studio_evidence_chunks;
CREATE POLICY evidence_chunks_no_client_insert ON public.brand_studio_evidence_chunks AS RESTRICTIVE
  FOR INSERT TO anon, authenticated WITH CHECK (false);
DROP POLICY IF EXISTS evidence_chunks_no_client_update ON public.brand_studio_evidence_chunks;
CREATE POLICY evidence_chunks_no_client_update ON public.brand_studio_evidence_chunks AS RESTRICTIVE
  FOR UPDATE TO anon, authenticated USING (false) WITH CHECK (false);
DROP POLICY IF EXISTS evidence_chunks_no_client_delete ON public.brand_studio_evidence_chunks;
CREATE POLICY evidence_chunks_no_client_delete ON public.brand_studio_evidence_chunks AS RESTRICTIVE
  FOR DELETE TO anon, authenticated USING (false);
REVOKE ALL ON FUNCTION public.brand_studio_store_evidence_chunk_v1(uuid,uuid,text,text,text) FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_store_evidence_chunk_v1(uuid,uuid,text,text,text) TO service_role;

-- The chunk lane's heartbeat renews the hard SQL lease through this RPC.
-- Never accept a caller's signals snapshot or resurrect an expired/released
-- lease. Only one JSON path is changed, using the current row under its lock.
CREATE OR REPLACE FUNCTION public.brand_studio_renew_evidence_lease_v1(
  p_pull_id uuid, p_user_id uuid, p_lease_token text, p_lease_ms integer
) RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path = '' AS $$
DECLARE
  stored_status text;
  stored_token text;
  stored_expiry text;
  previous_expiry timestamptz;
  locked_at timestamptz;
  next_expiry timestamptz;
BEGIN
  IF p_pull_id IS NULL OR p_user_id IS NULL OR p_lease_token IS NULL OR btrim(p_lease_token) = ''
    OR p_lease_ms IS NULL OR p_lease_ms NOT BETWEEN 30000 AND 600000 THEN
    RAISE EXCEPTION 'evidence_lease_invalid_scope_or_duration' USING ERRCODE = '22023';
  END IF;
  SELECT status, signals #>> '{queue,lease_token}', signals #>> '{queue,lease_expires_at}'
    INTO stored_status, stored_token, stored_expiry
    FROM public.brand_studio_pulls WHERE id = p_pull_id AND user_id = p_user_id FOR UPDATE;
  IF NOT FOUND OR stored_status IS DISTINCT FROM 'running' OR stored_token IS DISTINCT FROM p_lease_token THEN
    RAISE EXCEPTION 'evidence_lease_stale_token_or_owner' USING ERRCODE = '42501';
  END IF;
  BEGIN
    previous_expiry := stored_expiry::timestamptz;
  EXCEPTION WHEN invalid_datetime_format OR datetime_field_overflow THEN
    RAISE EXCEPTION 'evidence_lease_invalid_expiry' USING ERRCODE = '42501';
  END;
  locked_at := clock_timestamp();
  IF previous_expiry IS NULL OR NOT isfinite(previous_expiry) OR previous_expiry <= locked_at THEN
    RAISE EXCEPTION 'evidence_lease_expired' USING ERRCODE = '42501';
  END IF;
  next_expiry := locked_at + p_lease_ms * interval '1 millisecond';
  UPDATE public.brand_studio_pulls
    SET signals = jsonb_set(signals, '{queue,lease_expires_at}', to_jsonb(next_expiry), false)
    WHERE id = p_pull_id AND user_id = p_user_id;
  RETURN jsonb_build_object('pull_id', p_pull_id, 'user_id', p_user_id,
    'lease_token', p_lease_token, 'lease_expires_at', next_expiry);
END;
$$;
REVOKE ALL ON FUNCTION public.brand_studio_renew_evidence_lease_v1(uuid,uuid,text,integer) FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_renew_evidence_lease_v1(uuid,uuid,text,integer) TO service_role;

-- Read-only readiness probe for NEW RUN opt-in before reservation. This marker
-- becomes visible only when the whole migration commits. No row writes/money.
CREATE OR REPLACE FUNCTION public.brand_studio_evidence_storage_version_v1()
RETURNS jsonb LANGUAGE sql STABLE SECURITY INVOKER SET search_path = '' AS $$
  SELECT jsonb_build_object('schema_version', 'brand-studio-evidence-storage/v1');
$$;
REVOKE ALL ON FUNCTION public.brand_studio_evidence_storage_version_v1() FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_evidence_storage_version_v1() TO service_role;
COMMIT;