d3ro-voice/server/supabase/migrations/20260821000013_push_dispatch_limits.sql

BEGIN;

-- Transcription completion and a specific invitation are immutable events.
-- Keep the earliest reservation if a pre-release database already contains
-- duplicates, then enforce exactly-once dispatch across user/system actors.
WITH ranked AS (
  SELECT id, row_number() OVER (
    PARTITION BY event_type, resource_id
    ORDER BY created_at, id
  ) AS duplicate_rank
  FROM public.push_dispatch_attempts
  WHERE event_type IN ('transcription.completed', 'team.invite.created')
)
DELETE FROM public.push_dispatch_attempts AS attempt
USING ranked
WHERE attempt.id = ranked.id AND ranked.duplicate_rank > 1;

CREATE UNIQUE INDEX idx_push_dispatch_immutable_event_once
  ON public.push_dispatch_attempts(event_type, resource_id)
  WHERE event_type IN ('transcription.completed', 'team.invite.created');

CREATE OR REPLACE FUNCTION public.reserve_push_dispatch(
  push_event_type text,
  push_resource_id uuid
)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  current_user_id uuid := auth.uid();
  normalized_event_type text := lower(btrim(push_event_type));
  existing_attempt_id uuid;
  created_attempt_id uuid;
BEGIN
  IF current_user_id IS NULL THEN
    RAISE EXCEPTION 'authentication_required' USING ERRCODE = '42501';
  END IF;
  IF push_resource_id IS NULL OR normalized_event_type NOT IN (
    'transcription.completed',
    'team.invite.created',
    'billing.status.changed'
  ) THEN
    RAISE EXCEPTION 'invalid_push_event' USING ERRCODE = '22023';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended(current_user_id::text, 73048));
  PERFORM pg_advisory_xact_lock(
    hashtextextended(normalized_event_type || ':' || push_resource_id::text, 73051)
  );

  SELECT attempt.id INTO existing_attempt_id
  FROM public.push_dispatch_attempts AS attempt
  WHERE attempt.event_type = normalized_event_type
    AND attempt.resource_id = push_resource_id
    AND (
      normalized_event_type IN ('transcription.completed', 'team.invite.created')
      OR attempt.created_at > now() - interval '30 seconds'
    )
  ORDER BY attempt.created_at DESC
  LIMIT 1;
  IF existing_attempt_id IS NOT NULL THEN
    RETURN jsonb_build_object(
      'reserved', false,
      'duplicate', true,
      'attempt_id', existing_attempt_id
    );
  END IF;

  IF (
    SELECT count(*)
    FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'user'
      AND attempt.caller_id = current_user_id
      AND attempt.created_at > now() - interval '1 minute'
  ) >= 10 OR (
    SELECT count(*)
    FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'user'
      AND attempt.caller_id = current_user_id
      AND attempt.created_at > now() - interval '24 hours'
  ) >= 100 THEN
    RAISE EXCEPTION 'push_rate_limited' USING ERRCODE = '54000';
  END IF;

  INSERT INTO public.push_dispatch_attempts (
    caller_id, actor_kind, event_type, resource_id
  ) VALUES (
    current_user_id, 'user', normalized_event_type, push_resource_id
  )
  RETURNING id INTO created_attempt_id;
  RETURN jsonb_build_object(
    'reserved', true,
    'duplicate', false,
    'attempt_id', created_attempt_id
  );
END;
$$;

CREATE OR REPLACE FUNCTION public.reserve_system_push_dispatch(
  push_event_type text,
  push_resource_id uuid
)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  normalized_event_type text := lower(btrim(push_event_type));
  existing_attempt_id uuid;
  created_attempt_id uuid;
BEGIN
  IF auth.role() <> 'service_role' THEN
    RAISE EXCEPTION 'service_role_required' USING ERRCODE = '42501';
  END IF;
  IF push_resource_id IS NULL OR normalized_event_type NOT IN (
    'transcription.completed',
    'team.invite.created',
    'billing.status.changed'
  ) THEN
    RAISE EXCEPTION 'invalid_push_event' USING ERRCODE = '22023';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended('system-push-global', 73050));
  PERFORM pg_advisory_xact_lock(
    hashtextextended(normalized_event_type || ':' || push_resource_id::text, 73051)
  );

  SELECT attempt.id INTO existing_attempt_id
  FROM public.push_dispatch_attempts AS attempt
  WHERE attempt.event_type = normalized_event_type
    AND attempt.resource_id = push_resource_id
    AND (
      normalized_event_type IN ('transcription.completed', 'team.invite.created')
      OR attempt.created_at > now() - interval '30 seconds'
    )
  ORDER BY attempt.created_at DESC
  LIMIT 1;
  IF existing_attempt_id IS NOT NULL THEN
    RETURN jsonb_build_object(
      'reserved', false,
      'duplicate', true,
      'attempt_id', existing_attempt_id
    );
  END IF;

  IF (
    SELECT count(*)
    FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'system'
      AND attempt.created_at > now() - interval '1 minute'
  ) >= 500 OR (
    SELECT count(*)
    FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'system'
      AND attempt.created_at > now() - interval '24 hours'
  ) >= 10000 THEN
    RAISE EXCEPTION 'push_rate_limited' USING ERRCODE = '54000';
  END IF;

  INSERT INTO public.push_dispatch_attempts (
    caller_id, actor_kind, event_type, resource_id
  ) VALUES (
    NULL, 'system', normalized_event_type, push_resource_id
  )
  RETURNING id INTO created_attempt_id;
  RETURN jsonb_build_object(
    'reserved', true,
    'duplicate', false,
    'attempt_id', created_attempt_id
  );
END;
$$;

REVOKE ALL ON FUNCTION public.reserve_push_dispatch(text, uuid)
  FROM PUBLIC, anon;
GRANT EXECUTE ON FUNCTION public.reserve_push_dispatch(text, uuid)
  TO authenticated;
REVOKE ALL ON FUNCTION public.reserve_system_push_dispatch(text, uuid)
  FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.reserve_system_push_dispatch(text, uuid)
  TO service_role;

COMMIT;