supabase-brand-studio-notifications

-- 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;