Jyotisha/frontend/supabase/migrations/20260930040000_moderation_events.sql

-- Content moderation hits (2026-09-30, compliance round). Add-only.
-- One row per blocked or flagged input/output. No user or model text is
-- stored: only categories, rule ids / provider labels, route, side, model and
-- request ids. Service role only; operators read it through /admin.
create table if not exists public.moderation_events (
  id bigint generated always as identity primary key,
  created_at timestamptz not null default now(),
  user_id uuid references auth.users(id) on delete set null,
  side text not null check (side in ('input', 'output')),
  verdict text not null check (verdict in ('block', 'review')),
  route text not null check (char_length(route) between 1 and 64),
  categories text[] not null default '{}',
  rule_ids text[] not null default '{}',
  provider text not null check (provider in ('local', 'aliyun')),
  model_id text,
  request_id text
);

create index if not exists moderation_events_created_idx on public.moderation_events (created_at desc);
create index if not exists moderation_events_user_idx on public.moderation_events (user_id, created_at desc);

alter table public.moderation_events enable row level security;
revoke all on public.moderation_events from public, anon, authenticated;
grant select, insert on public.moderation_events to service_role;

-- Output moderation (same round): an answer whose text failed moderation is
-- replaced by a short notice and the reservation is released, exactly like
-- the small-talk free completion above it in history, but for the notice
-- (responseKind 'moderated', ≤ 200 chars). Service role only.
begin;
create or replace function public.complete_consultation_moderated(
  p_user_id uuid,
  p_request_id text,
  p_session_id uuid,
  p_response_message jsonb,
  p_actual_usage jsonb
)
returns table(success boolean, credits integer, error_code text)
language plpgsql
security definer
set search_path = ''
as $$
declare
  v_request public.consultation_requests%rowtype;
  v_res public.usage_reservations%rowtype;
  v_settlement record;
  v_balance integer;
begin
  if p_user_id is null or p_session_id is null or btrim(coalesce(p_request_id, '')) = '' then
    return query select false, null::integer, 'invalid_request'::text;
    return;
  end if;
  if jsonb_typeof(p_response_message) is distinct from 'object'
    or p_response_message->>'role' is distinct from 'assistant'
    or p_response_message->>'responseKind' is distinct from 'moderated'
    or jsonb_typeof(p_response_message->'text') is distinct from 'string'
    or btrim(coalesce(p_response_message->>'text', '')) = ''
    or length(p_response_message->>'text') > 200
    or (p_response_message - array['role', 'text', 'responseKind']) <> '{}'::jsonb then
    return query select false, null::integer, 'invalid_response_message'::text;
    return;
  end if;
  if jsonb_typeof(p_actual_usage) is distinct from 'object' then
    return query select false, null::integer, 'invalid_actual_usage'::text;
    return;
  end if;

  perform pg_advisory_xact_lock(hashtextextended(p_user_id::text || ':' || btrim(p_request_id), 0));
  select request.* into v_request
  from public.consultation_requests as request
  where request.user_id = p_user_id
    and request.request_id = btrim(p_request_id)
    and request.session_id = p_session_id
  for update;
  if not found then
    return query select false, null::integer, 'request_missing'::text;
    return;
  end if;
  if v_request.status = 'cancelled' then
    return query select false, null::integer, 'request_cancelled'::text;
    return;
  end if;
  if v_request.status = 'completed' then
    select profile.credits into v_balance from public.profiles as profile where profile.id = p_user_id;
    return query select coalesce(v_request.response_message = p_response_message, false), v_balance,
      case when v_request.response_message = p_response_message then null::text else 'response_conflict'::text end;
    return; -- Never refund a previously completed paid consultation.
  end if;
  if v_request.status <> 'reserved' then
    return query select false, null::integer, 'invalid_request_status'::text;
    return;
  end if;

  select reservation.* into v_res from public.usage_reservations as reservation
  where reservation.user_id = p_user_id and reservation.request_id = btrim(p_request_id)
  for update;
  if not found or v_res.status <> 'reserved' or v_res.feature_key <> 'chat.standard' then
    return query select false, null::integer, 'invalid_reservation'::text;
    return;
  end if;
  select profile.credits into v_balance from public.profiles as profile
  where profile.id = p_user_id for update;
  if not found then
    return query select false, null::integer, 'profile_missing'::text;
    return;
  end if;

  update public.chat_sessions as session
  set messages = session.messages || jsonb_build_array(p_response_message), updated_at = clock_timestamp()
  where session.id = p_session_id and session.user_id = p_user_id and session.session_type = 'consultation';
  if not found then
    return query select false, null::integer, 'session_missing'::text;
    return;
  end if;

  -- complete_usage is the sole cost ledger writer; it does not debit credits.
  -- Keep cost even though the reservation is released below (also frees subscription quota).
  select * into v_settlement from public.complete_usage(
    p_user_id, btrim(p_request_id),
    p_actual_usage || jsonb_build_object('metadata', coalesce(p_actual_usage->'metadata', '{}'::jsonb)
      || jsonb_build_object('responseKind', 'moderated', 'freeCompletion', true))
  );
  if not coalesce(v_settlement.success, false) then
    raise exception 'consultation_moderated_usage_settlement_failed:%', coalesce(v_settlement.error_code, 'unknown');
  end if;

  -- Same refund amount, balance lock and transaction identity as release_usage.
  -- Only reachable from a reserved request and reserved usage row under the shared lock.
  if v_res.source = 'credits' and v_res.credit_amount > 0 then
    update public.profiles as profile
    set credits = profile.credits + v_res.credit_amount, updated_at = clock_timestamp()
    where profile.id = p_user_id returning profile.credits into v_balance;
    insert into public.credit_transactions(user_id, transaction_type, amount, balance_after, request_id, model)
    values(p_user_id, 'refund', v_res.credit_amount, v_balance, v_res.request_id, v_res.requested_model_id);
    -- No ON CONFLICT: an inconsistent prior refund must roll back everything, never double-credit.
  end if;
  update public.usage_reservations
  set status = 'released', released_at = clock_timestamp(), release_reason = 'consultation_output_moderated'
  where id = v_res.id;
  update public.consultation_requests
  set status = 'completed', response_message = p_response_message, updated_at = clock_timestamp()
  where user_id = p_user_id and request_id = btrim(p_request_id);

  return query select true, v_balance, null::text;
end;
$$;

revoke all on function public.complete_consultation_moderated(uuid, text, uuid, jsonb, jsonb)
  from public, anon, authenticated;
grant execute on function public.complete_consultation_moderated(uuid, text, uuid, jsonb, jsonb)
  to service_role;
commit;