supabase-brand-studio-monitoring-20261023

-- Monitoring release 2026-10-23. Apply once; reapplication is safe.
-- Requires existing Auth, brand and credit ledger tables. No rows are deleted or rewritten.
BEGIN;
SET LOCAL lock_timeout = '5s';
SET LOCAL statement_timeout = '30s';

-- DRAFT: owner-approved weekly ten-newest review samples. Apply only with release coordination.
SET LOCAL lock_timeout = '5s';
SET LOCAL statement_timeout = '30s';

CREATE TABLE IF NOT EXISTS public.brand_studio_review_watches (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
  brand_id uuid NOT NULL,
  domain text NOT NULL CHECK (char_length(domain) BETWEEN 1 AND 253),
  cadence text NOT NULL DEFAULT 'weekly' CHECK (cadence IN ('weekly')),
  status text NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'paused')),
  paused_reason text CHECK (paused_reason IS NULL OR paused_reason IN ('owner', 'failing', 'credits', 'limit', 'review')),
  profile jsonb NOT NULL CHECK (jsonb_typeof(profile) = 'object' AND octet_length(profile::text) <= 4096 AND profile @> '{"confirmed":true}'::jsonb),
  market jsonb NOT NULL CHECK (jsonb_typeof(market) = 'object' AND market @> '{"confirmed":true}'::jsonb),
  plan_hash text NOT NULL CHECK (plan_hash ~ '^[a-f0-9]{64}$'),
  max_cost_usd numeric(10, 6) NOT NULL CHECK (max_cost_usd > 0 AND max_cost_usd <= 3),
  credits_per_run integer NOT NULL CHECK (credits_per_run BETWEEN 1 AND 3),
  approved_at timestamptz NOT NULL,
  anchor_at timestamptz NOT NULL,
  next_check_at timestamptz NOT NULL,
  last_check_at timestamptz,
  last_outcome text CHECK (last_outcome IS NULL OR last_outcome IN ('started', 'refused', 'failed')),
  last_pull_id uuid,
  last_reason text CHECK (last_reason IS NULL OR char_length(last_reason) <= 200),
  consecutive_failures integer NOT NULL DEFAULT 0 CHECK (consecutive_failures BETWEEN 0 AND 1000),
  lease_until timestamptz,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now(),
  CONSTRAINT brand_studio_review_watches_brand_owner_fk
    FOREIGN KEY (brand_id, user_id) REFERENCES public.brand_studio_brands (id, user_id) ON DELETE CASCADE,
  CONSTRAINT brand_studio_review_watches_one_per_brand UNIQUE (user_id, brand_id),
  CONSTRAINT brand_studio_review_watches_id_owner UNIQUE (id, user_id)
);

CREATE INDEX IF NOT EXISTS brand_studio_review_watches_due_idx
  ON public.brand_studio_review_watches (next_check_at) WHERE status = 'active';

CREATE TABLE IF NOT EXISTS public.brand_studio_review_watch_runs (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  watch_id uuid NOT NULL,
  user_id uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
  occurrence_at timestamptz NOT NULL,
  status text NOT NULL CHECK (status IN ('running', 'started', 'refused', 'failed')),
  pull_id uuid,
  plan_hash text CHECK (plan_hash IS NULL OR plan_hash ~ '^[a-f0-9]{64}$'),
  max_cost_usd numeric(10, 6),
  reason text CHECK (reason IS NULL OR char_length(reason) <= 200),
  started_at timestamptz NOT NULL DEFAULT now(),
  finished_at timestamptz,
  CONSTRAINT brand_studio_review_watch_runs_watch_owner_fk
    FOREIGN KEY (watch_id, user_id) REFERENCES public.brand_studio_review_watches (id, user_id) ON DELETE CASCADE,
  CONSTRAINT brand_studio_review_watch_runs_one_per_occurrence UNIQUE (watch_id, occurrence_at)
);

ALTER TABLE public.brand_studio_review_watch_runs
  ADD COLUMN IF NOT EXISTS sample_scope jsonb NOT NULL DEFAULT '{"review_limit":10,"sort":"newest","overlap_possible":true,"interval_complete":false}'::jsonb,
  ADD COLUMN IF NOT EXISTS listing jsonb,
  ADD COLUMN IF NOT EXISTS market jsonb;

CREATE INDEX IF NOT EXISTS brand_studio_review_watch_runs_user_started_idx
  ON public.brand_studio_review_watch_runs (user_id, started_at DESC);

ALTER TABLE public.brand_studio_review_watches ENABLE ROW LEVEL SECURITY;
ALTER TABLE public.brand_studio_review_watch_runs ENABLE ROW LEVEL SECURITY;
REVOKE ALL ON TABLE public.brand_studio_review_watches FROM PUBLIC, anon, authenticated;
REVOKE ALL ON TABLE public.brand_studio_review_watch_runs FROM PUBLIC, anon, authenticated;
DROP POLICY IF EXISTS brand_studio_review_watches_select_own ON public.brand_studio_review_watches;
CREATE POLICY brand_studio_review_watches_select_own ON public.brand_studio_review_watches
  FOR SELECT TO authenticated USING (user_id = (SELECT auth.uid()));
DROP POLICY IF EXISTS brand_studio_review_watch_runs_select_own ON public.brand_studio_review_watch_runs;
CREATE POLICY brand_studio_review_watch_runs_select_own ON public.brand_studio_review_watch_runs
  FOR SELECT TO authenticated USING (user_id = (SELECT auth.uid()));
GRANT SELECT ON TABLE public.brand_studio_review_watches TO authenticated;
GRANT SELECT ON TABLE public.brand_studio_review_watch_runs TO authenticated;
-- New schedules must use the owner-fenced creation function, including old workers.
REVOKE INSERT ON TABLE public.brand_studio_review_watches FROM service_role;
GRANT SELECT, UPDATE, DELETE ON TABLE public.brand_studio_review_watches TO service_role;
GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE public.brand_studio_review_watch_runs TO service_role;

-- Serialize creation for one owner. A REST count followed by INSERT races at nine.
-- Lock the durable owner row so even different brands share the same fence.
CREATE OR REPLACE FUNCTION public.brand_studio_create_review_watch(
  p_user_id uuid, p_brand_id uuid, p_domain text, p_watch jsonb
) RETURNS SETOF public.brand_studio_review_watches
LANGUAGE plpgsql SECURITY DEFINER SET search_path = '' AS $$
BEGIN
  PERFORM 1 FROM auth.users WHERE id = p_user_id FOR UPDATE;
  IF NOT FOUND THEN
    RAISE EXCEPTION USING ERRCODE = '23503', MESSAGE = 'review_watch_owner_not_found';
  END IF;
  IF NOT EXISTS (SELECT 1 FROM public.brand_studio_brands
    WHERE id = p_brand_id AND user_id = p_user_id AND domain = p_domain) THEN
    RAISE EXCEPTION USING ERRCODE = '23503', MESSAGE = 'review_watch_brand_not_owned';
  END IF;
  IF (SELECT count(*) FROM public.brand_studio_review_watches WHERE user_id = p_user_id) >= 10 THEN
    RAISE EXCEPTION USING ERRCODE = 'P0010', MESSAGE = 'schedule_limit_reached';
  END IF;
  RETURN QUERY INSERT INTO public.brand_studio_review_watches (
    user_id, brand_id, domain, cadence, status, profile, market, plan_hash,
    max_cost_usd, credits_per_run, approved_at, anchor_at, next_check_at, created_at, updated_at
  ) VALUES (
    p_user_id, p_brand_id, p_domain, 'weekly', 'active', p_watch->'profile', p_watch->'market',
    p_watch->>'plan_hash', (p_watch->>'max_cost_usd')::numeric, (p_watch->>'credits_per_run')::integer,
    (p_watch->>'approved_at')::timestamptz, (p_watch->>'anchor_at')::timestamptz,
    (p_watch->>'next_check_at')::timestamptz, (p_watch->>'updated_at')::timestamptz,
    (p_watch->>'updated_at')::timestamptz
  ) RETURNING *;
END;
$$;
REVOKE ALL ON FUNCTION public.brand_studio_create_review_watch(uuid, uuid, text, jsonb) FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_create_review_watch(uuid, uuid, text, jsonb) TO service_role;


-- Prepared only: parent applies separately. Email/payload are private service-role data.
create table if not exists public.brand_studio_notification_preferences (
  user_id uuid primary key references auth.users(id) on delete cascade,
  recipient text not null,
  enabled boolean not null default false,
  version uuid not null default gen_random_uuid(),
  suppression_reason text,
  updated_at timestamptz not null default now()
);
create table if not exists public.brand_studio_notification_suppressions (
  recipient text primary key, reason text not null, suppressed_at timestamptz not null default now()
);
create table if not exists public.brand_studio_notification_outbox (
  id uuid primary key,
  user_id uuid not null references auth.users(id) on delete cascade,
  preference_version uuid not null,
  recipient text not null,
  status text not null check (status in ('pending','sending','retry','unknown','accepted','delivered','suppressed','failed')),
  idempotency_key text not null unique,
  payload jsonb not null,
  payload_sha256 text not null check (payload_sha256 ~ '^[a-f0-9]{64}$'),
  provider_id text,
  reason text,
  created_at timestamptz not null default now(),
  first_attempt_at timestamptz,
  accepted_at timestamptz,
  delivered_at timestamptz,
  attempts integer not null default 0,
  lease_token uuid,
  lease_until timestamptz,
  next_attempt_at timestamptz not null default now()
);
alter table public.brand_studio_notification_outbox drop constraint if exists brand_studio_notification_outbox_status_check;
alter table public.brand_studio_notification_outbox add constraint brand_studio_notification_outbox_status_check
  check (status in ('pending','sending','retry','unknown','accepted','delivered','suppressed','failed'));
alter table public.brand_studio_notification_preferences
  add column if not exists next_queue_at timestamptz not null default now(),
  add column if not exists queue_lease uuid,
  add column if not exists queue_lease_until timestamptz;
create index if not exists brand_studio_notification_queue_due_idx on public.brand_studio_notification_preferences(next_queue_at,user_id) where enabled and suppression_reason is null;
create index if not exists brand_studio_notification_due_idx on public.brand_studio_notification_outbox(next_attempt_at) where status in ('pending','retry','sending','accepted','delivered');
alter table public.brand_studio_notification_suppressions enable row level security;
revoke all on public.brand_studio_notification_suppressions from public, anon, authenticated;
grant select on public.brand_studio_notification_suppressions to service_role;
insert into public.brand_studio_notification_suppressions(recipient,reason)
  select lower(recipient),suppression_reason from public.brand_studio_notification_preferences where suppression_reason is not null on conflict do nothing;
alter table public.brand_studio_notification_preferences enable row level security;
alter table public.brand_studio_notification_outbox enable row level security;
-- No client policies: routes enforce session owner and return redacted projections.
revoke all on public.brand_studio_notification_preferences, public.brand_studio_notification_outbox from public, anon, authenticated;
grant all on public.brand_studio_notification_preferences, public.brand_studio_notification_outbox to service_role;

create or replace function public.brand_studio_notification_preference(p_user_id uuid, p_email text, p_enabled boolean)
returns setof public.brand_studio_notification_preferences language plpgsql security definer set search_path = public, pg_temp as $$
begin
  if not exists (select 1 from auth.users where id=p_user_id and email=p_email and email_confirmed_at is not null) then raise exception 'verified_email_required'; end if;
  insert into public.brand_studio_notification_preferences(user_id, recipient, enabled) values (p_user_id,p_email,false) on conflict do nothing;
  perform 1 from public.brand_studio_notification_preferences where user_id=p_user_id for update;
  if p_enabled and (exists (select 1 from public.brand_studio_notification_suppressions where recipient=lower(p_email)) or exists (select 1 from public.brand_studio_notification_preferences where user_id=p_user_id and suppression_reason is not null and recipient=p_email)) then raise exception 'provider_suppression_requires_review'; end if;
  return query update public.brand_studio_notification_preferences
    set recipient=p_email, enabled=p_enabled,
      version=case when recipient is distinct from p_email or enabled is distinct from p_enabled then gen_random_uuid() else version end,
      updated_at=case when recipient is distinct from p_email or enabled is distinct from p_enabled then now() else updated_at end,
      next_queue_at=case when recipient is distinct from p_email or enabled is distinct from p_enabled then now() else next_queue_at end,
      queue_lease=case when recipient is distinct from p_email or enabled is distinct from p_enabled then null else queue_lease end,
      queue_lease_until=case when recipient is distinct from p_email or enabled is distinct from p_enabled then null else queue_lease_until end,
      suppression_reason=case when recipient<>p_email then null else suppression_reason end
    where user_id=p_user_id returning *;
  if not p_enabled then
    update public.brand_studio_notification_outbox set status='suppressed',reason='consent_not_current',lease_token=null,lease_until=null
      where user_id=p_user_id and provider_id is null and status in ('pending','retry');
  end if;
end $$;
create or replace function public.brand_studio_notification_queue_claim(p_limit integer)
returns setof public.brand_studio_notification_preferences language plpgsql security definer set search_path = public, pg_temp as $$
begin
  if p_limit is null or p_limit<1 or p_limit>2 then raise exception 'bounded_batch_required'; end if;
  return query with due as (
    select p.user_id from public.brand_studio_notification_preferences p
    join auth.users u on u.id=p.user_id and u.email=p.recipient and u.email_confirmed_at is not null
    where p.enabled and p.suppression_reason is null and p.next_queue_at<=now() and (p.queue_lease_until is null or p.queue_lease_until<now())
    order by p.next_queue_at,p.user_id for update of p skip locked limit p_limit
  ) update public.brand_studio_notification_preferences p set queue_lease=gen_random_uuid(),queue_lease_until=now()+interval '2 minutes'
    from due where p.user_id=due.user_id returning p.*;
end $$;
create or replace function public.brand_studio_notification_queue_finish(p_user_id uuid,p_version uuid,p_lease uuid,p_queued boolean)
returns boolean language plpgsql security definer set search_path = public, pg_temp as $$
begin
  update public.brand_studio_notification_preferences set queue_lease=null,queue_lease_until=null,
    next_queue_at=case when p_queued then now()+interval '7 days' else now()+interval '5 minutes' end
    where user_id=p_user_id and version=p_version and queue_lease=p_lease;
  return found;
end $$;
create or replace function public.brand_studio_notification_unsubscribe(p_user_id uuid,p_version uuid)
returns boolean language plpgsql security definer set search_path = public, pg_temp as $$
begin
  -- Old messages may only disable their own consent version, never later re-opt-in.
  update public.brand_studio_notification_preferences set enabled=false,updated_at=now() where user_id=p_user_id and version=p_version;
  update public.brand_studio_notification_outbox set status='suppressed',reason='consent_not_current',lease_token=null,lease_until=null
    where user_id=p_user_id and preference_version=p_version and provider_id is null and status in ('pending','retry');
  return true;
end $$;
create or replace function public.brand_studio_notification_enqueue(p_entry jsonb)
returns boolean language plpgsql security definer set search_path = public, pg_temp as $$
declare pref public.brand_studio_notification_preferences;
begin
  select * into pref from public.brand_studio_notification_preferences where user_id=(p_entry->>'user_id')::uuid for update;
  if not found or not pref.enabled or pref.suppression_reason is not null or pref.version<>(p_entry->>'preference_version')::uuid or pref.recipient<>p_entry->>'recipient' then raise exception 'consent_not_current'; end if;
  insert into public.brand_studio_notification_outbox(id,user_id,preference_version,recipient,status,idempotency_key,payload,payload_sha256,created_at)
    values ((p_entry->>'id')::uuid,pref.user_id,pref.version,pref.recipient,'pending',p_entry->>'idempotency_key',p_entry->'payload',p_entry->>'payload_sha256',now()) on conflict (id) do nothing;
  return true;
end $$;
create or replace function public.brand_studio_notification_claim(p_limit integer)
returns setof public.brand_studio_notification_outbox language plpgsql security definer set search_path = public, pg_temp as $$
begin
  if p_limit is null or p_limit<1 or p_limit>2 then raise exception 'bounded_batch_required'; end if;
  update public.brand_studio_notification_outbox set status='unknown',reason='idempotency_or_attempt_limit_requires_reconciliation',lease_token=null,lease_until=null
    where provider_id is null and status in ('pending','retry','sending') and (lease_until is null or lease_until<now())
      and (first_attempt_at<=now()-interval '23 hours' or attempts>=5);
  update public.brand_studio_notification_outbox set status='unknown',reason='delivery_status_requires_reconciliation',lease_token=null,lease_until=null
    where provider_id is not null and delivered_at is null and status in ('accepted','sending') and accepted_at<now()-interval '7 days' and (lease_until is null or lease_until<now());
  return query with due as (
    select id from public.brand_studio_notification_outbox
    where status in ('pending','retry','sending','accepted','delivered') and next_attempt_at<=now() and (lease_until is null or lease_until<now())
      and (delivered_at is null or delivered_at>=now()-interval '30 days')
    order by case when provider_id is null then 0 when delivered_at is null then 1 else 2 end,next_attempt_at,id for update skip locked limit p_limit
  ) update public.brand_studio_notification_outbox o set status='sending',lease_token=gen_random_uuid(),lease_until=now()+interval '2 minutes',
      first_attempt_at=case when o.provider_id is null then coalesce(o.first_attempt_at,now()) else o.first_attempt_at end,
      attempts=case when o.provider_id is null then o.attempts+1 else o.attempts end
    from due where o.id=due.id returning o.*;
end $$;
create or replace function public.brand_studio_notification_finish(p_id uuid,p_lease uuid,p_status text,p_reason text,p_provider_id text)
returns boolean language plpgsql security definer set search_path = public, pg_temp as $$
declare saved public.brand_studio_notification_outbox;
begin
  if p_status not in ('retry','unknown','accepted','delivered','suppressed','failed') or length(p_reason)>100 then raise exception 'invalid_notification_result'; end if;
  select * into saved from public.brand_studio_notification_outbox where id=p_id and lease_token=p_lease and status='sending' for update;
  if not found then return false; end if;
  if p_status in ('accepted','delivered') and p_provider_id is null then raise exception 'provider_identity_required'; end if;
  update public.brand_studio_notification_outbox set status=p_status,reason=p_reason,provider_id=coalesce(saved.provider_id,p_provider_id),lease_token=null,lease_until=null,
    accepted_at=case when p_status in ('accepted','delivered') then coalesce(saved.accepted_at,now()) else saved.accepted_at end,
    delivered_at=case when p_status='delivered' then coalesce(saved.delivered_at,now()) else saved.delivered_at end,
    next_attempt_at=now()+case when p_status='delivered' then interval '1 day' when p_status='accepted' then interval '1 hour' else interval '5 minutes' end where id=p_id;
  if p_status in ('suppressed','failed') and p_reason in ('provider_bounced','provider_complained','provider_failed','provider_suppressed') then
    insert into public.brand_studio_notification_suppressions(recipient,reason) values(lower(saved.recipient),p_reason) on conflict(recipient) do update set reason=excluded.reason,suppressed_at=now();
    update public.brand_studio_notification_preferences set enabled=false,suppression_reason=p_reason,updated_at=now() where lower(recipient)=lower(saved.recipient);
  end if;
  return true;
end $$;
revoke all on function public.brand_studio_notification_preference(uuid,text,boolean),public.brand_studio_notification_unsubscribe(uuid,uuid),public.brand_studio_notification_enqueue(jsonb),public.brand_studio_notification_claim(integer),public.brand_studio_notification_finish(uuid,uuid,text,text,text) from public,anon,authenticated;
grant execute on function public.brand_studio_notification_preference(uuid,text,boolean),public.brand_studio_notification_unsubscribe(uuid,uuid),public.brand_studio_notification_enqueue(jsonb),public.brand_studio_notification_claim(integer),public.brand_studio_notification_finish(uuid,uuid,text,text,text) to service_role;
revoke all on function public.brand_studio_notification_queue_claim(integer),public.brand_studio_notification_queue_finish(uuid,uuid,uuid,boolean) from public,anon,authenticated;
grant execute on function public.brand_studio_notification_queue_claim(integer),public.brand_studio_notification_queue_finish(uuid,uuid,uuid,boolean) to service_role;

-- Additive shared scheduler lease and bounded latest health, service access only.
CREATE TABLE IF NOT EXISTS public.brand_studio_monitoring_sweeps (
  lane text PRIMARY KEY CHECK (lane IN ('basic','search','social','review','notification')),
  holder text,
  lease_token uuid,
  lease_until timestamptz,
  last_run_at timestamptz,
  last_finished_at timestamptz,
  last_skipped_at timestamptz,
  last_skipped_reason text,
  overlap_skips bigint NOT NULL DEFAULT 0,
  last_result jsonb NOT NULL DEFAULT '{}'::jsonb
);
ALTER TABLE public.brand_studio_monitoring_sweeps ENABLE ROW LEVEL SECURITY;
ALTER TABLE public.brand_studio_monitoring_sweeps
  ADD COLUMN IF NOT EXISTS last_skipped_at timestamptz,
  ADD COLUMN IF NOT EXISTS last_skipped_reason text,
  ADD COLUMN IF NOT EXISTS overlap_skips bigint NOT NULL DEFAULT 0;
REVOKE ALL ON public.brand_studio_monitoring_sweeps FROM PUBLIC, anon, authenticated;
GRANT SELECT ON public.brand_studio_monitoring_sweeps TO service_role;

CREATE OR REPLACE FUNCTION public.brand_studio_monitoring_sweep_claim(p_lane text,p_holder text)
RETURNS SETOF public.brand_studio_monitoring_sweeps
LANGUAGE plpgsql SECURITY DEFINER SET search_path = '' AS $$
BEGIN
  IF p_lane NOT IN ('basic','search','social','review','notification') OR length(p_holder)>80 THEN
    RAISE EXCEPTION 'invalid_sweep';
  END IF;
  INSERT INTO public.brand_studio_monitoring_sweeps(lane) VALUES(p_lane) ON CONFLICT DO NOTHING;
  RETURN QUERY UPDATE public.brand_studio_monitoring_sweeps
    SET holder=p_holder,lease_token=gen_random_uuid(),lease_until=now()+interval '10 minutes',last_run_at=now()
    WHERE lane=p_lane AND (lease_until IS NULL OR lease_until<now()) RETURNING *;
  IF NOT FOUND THEN
    UPDATE public.brand_studio_monitoring_sweeps SET last_skipped_at=now(),last_skipped_reason='sweep_lease_busy',overlap_skips=overlap_skips+1 WHERE lane=p_lane;
  END IF;
END $$;

CREATE OR REPLACE FUNCTION public.brand_studio_monitoring_sweep_finish(p_lane text,p_lease uuid,p_result jsonb)
RETURNS boolean LANGUAGE plpgsql SECURITY DEFINER SET search_path = '' AS $$
BEGIN
  IF jsonb_typeof(p_result)<>'object' OR octet_length(p_result::text)>2048 THEN RAISE EXCEPTION 'invalid_sweep_result'; END IF;
  UPDATE public.brand_studio_monitoring_sweeps
    SET lease_token=NULL,lease_until=NULL,last_finished_at=now(),last_result=p_result
    WHERE lane=p_lane AND lease_token=p_lease;
  RETURN FOUND;
END $$;
REVOKE ALL ON FUNCTION public.brand_studio_monitoring_sweep_claim(text,text),public.brand_studio_monitoring_sweep_finish(text,uuid,jsonb) FROM PUBLIC,anon,authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_monitoring_sweep_claim(text,text),public.brand_studio_monitoring_sweep_finish(text,uuid,jsonb) TO service_role;

-- Preserve the complete atomic ledger read and expose the immutable metadata
-- needed to distinguish an actual signup grant from configured beta copy.
CREATE OR REPLACE FUNCTION public.brand_studio_ledger_snapshot(p_user_id uuid)
RETURNS jsonb LANGUAGE sql STABLE SECURITY INVOKER SET search_path = '' AS $$
  WITH rows AS MATERIALIZED (
    SELECT kind,amount,pull_id,user_id,idempotency_key,created_at
    FROM public.brand_studio_credit_ledger
    WHERE user_id=p_user_id ORDER BY id LIMIT 100001
  )
  SELECT CASE WHEN count(*)>100000 THEN jsonb_build_object('error','ledger_limit_exceeded')
    ELSE jsonb_build_object('entries',coalesce(jsonb_agg(to_jsonb(rows)), '[]'::jsonb),
      'row_count',count(*),'snapshot_at',statement_timestamp()) END FROM rows;
$$;
REVOKE ALL ON FUNCTION public.brand_studio_ledger_snapshot(uuid) FROM PUBLIC,anon,authenticated;
GRANT EXECUTE ON FUNCTION public.brand_studio_ledger_snapshot(uuid) TO service_role;

COMMIT;