d3ro-voice/server/supabase/migrations/20260821000016_transactional_push_enqueue.sql

BEGIN;

-- Business state and notification enqueue must commit together. Edge calls
-- made after a commit can be lost if the worker crashes between the two.
CREATE OR REPLACE FUNCTION public.enqueue_immutable_push_event(
  queued_event_type text,
  queued_resource_id uuid
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
BEGIN
  IF queued_event_type NOT IN ('transcription.completed', 'team.invite.created')
     OR queued_resource_id IS NULL THEN
    RAISE EXCEPTION 'invalid_push_event' USING ERRCODE = '22023';
  END IF;
  INSERT INTO public.push_dispatch_attempts (
    caller_id, actor_kind, event_type, resource_id, status,
    attempt_count, next_retry_at
  ) VALUES (
    NULL, 'system', queued_event_type, queued_resource_id, 'pending',
    0, now()
  )
  ON CONFLICT DO NOTHING;
END;
$$;

CREATE OR REPLACE FUNCTION public.enqueue_billing_push_event(
  queued_subscription_id uuid
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
BEGIN
  IF queued_subscription_id IS NULL THEN
    RAISE EXCEPTION 'invalid_push_event' USING ERRCODE = '22023';
  END IF;
  PERFORM pg_advisory_xact_lock(
    hashtextextended('billing.status.changed:' || queued_subscription_id::text, 73052)
  );
  IF EXISTS (
    SELECT 1 FROM public.push_dispatch_attempts
    WHERE event_type = 'billing.status.changed'
      AND resource_id = queued_subscription_id
      AND created_at > now() - interval '30 seconds'
  ) THEN
    RETURN;
  END IF;
  INSERT INTO public.push_dispatch_attempts (
    caller_id, actor_kind, event_type, resource_id, status,
    attempt_count, next_retry_at
  ) VALUES (
    NULL, 'system', 'billing.status.changed', queued_subscription_id,
    'pending', 0, now()
  );
END;
$$;

CREATE OR REPLACE FUNCTION public.enqueue_processing_job_push()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
BEGIN
  IF NEW.kind = 'transcription'
     AND NEW.status = 'succeeded'
     AND NEW.history_id IS NOT NULL
     AND OLD.status IS DISTINCT FROM 'succeeded'
     AND EXISTS (
       SELECT 1 FROM public.history
       WHERE id = NEW.history_id AND user_id = NEW.user_id AND status = 'completed'
     ) THEN
    PERFORM public.enqueue_immutable_push_event(
      'transcription.completed', NEW.history_id
    );
  END IF;
  RETURN NEW;
END;
$$;

CREATE OR REPLACE FUNCTION public.enqueue_history_after_worker_push()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
BEGIN
  IF NEW.status = 'completed'
     AND OLD.status IS DISTINCT FROM 'completed'
     AND EXISTS (
       SELECT 1 FROM public.processing_jobs
       WHERE history_id = NEW.id
         AND user_id = NEW.user_id
         AND kind = 'transcription'
         AND status = 'succeeded'
     ) THEN
    PERFORM public.enqueue_immutable_push_event(
      'transcription.completed', NEW.id
    );
  END IF;
  RETURN NEW;
END;
$$;

CREATE OR REPLACE FUNCTION public.enqueue_team_invite_push()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
BEGIN
  PERFORM public.enqueue_immutable_push_event('team.invite.created', NEW.id);
  RETURN NEW;
END;
$$;

CREATE OR REPLACE FUNCTION public.enqueue_subscription_push()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
BEGIN
  IF ROW(
    OLD.tier, OLD.status, OLD.provider, OLD.current_period_end,
    OLD.cancel_at, OLD.auto_renewing
  ) IS DISTINCT FROM ROW(
    NEW.tier, NEW.status, NEW.provider, NEW.current_period_end,
    NEW.cancel_at, NEW.auto_renewing
  ) THEN
    PERFORM public.enqueue_billing_push_event(NEW.id);
  END IF;
  RETURN NEW;
END;
$$;

CREATE TRIGGER enqueue_processing_job_push_after_success
  AFTER UPDATE OF status ON public.processing_jobs
  FOR EACH ROW EXECUTE FUNCTION public.enqueue_processing_job_push();

CREATE TRIGGER enqueue_history_push_after_worker_success
  AFTER UPDATE OF status ON public.history
  FOR EACH ROW EXECUTE FUNCTION public.enqueue_history_after_worker_push();

CREATE TRIGGER enqueue_team_invite_push_after_insert
  AFTER INSERT ON public.team_invites
  FOR EACH ROW EXECUTE FUNCTION public.enqueue_team_invite_push();

CREATE TRIGGER enqueue_subscription_push_after_change
  AFTER UPDATE OF tier, status, provider, current_period_end, cancel_at, auto_renewing
  ON public.subscriptions
  FOR EACH ROW EXECUTE FUNCTION public.enqueue_subscription_push();

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

COMMIT;