Remove receive alias and fix generic RPC span attributes

Author: onurtemizkan

dev-packages/browser-integration-tests/suites/integrations/supabase/generic-rpc/test.ts

@@ -50,8 +50,7 @@ sentryTest(
         'sentry.op': 'db',
         'sentry.origin': 'auto.db.supabase',
         'db.system': 'postgresql',
-        'db.operation': 'insert',
-        'db.table': 'my_custom_function',
+        'db.operation': 'rpc',
         'db.params': { param1: 'value1' },
       }),
     });

dev-packages/e2e-tests/test-applications/supabase-nextjs/pages/api/queue/batch-flow.ts

@@ -29,22 +29,22 @@ export default async function handler(req: NextApiRequest, res: NextApiResponse)
     return res.status(500).json({ error: `Send batch failed: ${sendError.message}` });
   }
 
-  const { data: receiveData, error: receiveError } = await supabaseClient.rpc('receive', {
+  const { data: readData, error: readError } = await supabaseClient.rpc('read', {
     queue_name: 'batch-flow-queue',
-    vt: 30,
-    qty: 3,
+    sleep_seconds: 30,
+    n: 3,
   });
 
-  if (receiveError) {
-    return res.status(500).json({ error: `Receive failed: ${receiveError.message}` });
+  if (readError) {
+    return res.status(500).json({ error: `Read failed: ${readError.message}` });
   }
 
-  const processedMessages = receiveData?.map((msg: Record<string, unknown>) => ({
+  const processedMessages = readData?.map((msg: Record<string, unknown>) => ({
     messageId: msg.msg_id,
     message: msg.message,
   }));
 
-  const messageIds = receiveData?.map((msg: Record<string, unknown>) => msg.msg_id).filter(Boolean);
+  const messageIds = readData?.map((msg: Record<string, unknown>) => msg.msg_id).filter(Boolean);
   if (messageIds && messageIds.length > 0) {
     const { error: archiveError } = await supabaseClient.rpc('archive', {
       queue_name: 'batch-flow-queue',
@@ -61,7 +61,7 @@ export default async function handler(req: NextApiRequest, res: NextApiResponse)
     batchSize: 3,
     produced: { messageIds: sendData },
     consumed: {
-      count: receiveData?.length || 0,
+      count: readData?.length || 0,
       messages: processedMessages,
     },
   });

dev-packages/e2e-tests/test-applications/supabase-nextjs/pages/api/queue/producer-consumer-flow.ts

@@ -17,17 +17,17 @@ export default async function handler(req: NextApiRequest, res: NextApiResponse)
     return res.status(500).json({ error: `Send failed: ${sendError.message}` });
   }
 
-  const { data: receiveData, error: receiveError } = await supabaseClient.rpc('receive', {
+  const { data: readData, error: readError } = await supabaseClient.rpc('read', {
     queue_name: 'e2e-flow-queue',
-    vt: 30,
-    qty: 1,
+    sleep_seconds: 30,
+    n: 1,
   });
 
-  if (receiveError) {
-    return res.status(500).json({ error: `Receive failed: ${receiveError.message}` });
+  if (readError) {
+    return res.status(500).json({ error: `Read failed: ${readError.message}` });
   }
 
-  const processedMessage = receiveData?.[0];
+  const processedMessage = readData?.[0];
 
   if (processedMessage?.msg_id) {
     const { error: archiveError } = await supabaseClient.rpc('archive', {

dev-packages/e2e-tests/test-applications/supabase-nextjs/supabase/migrations/20250515080602_enable-queues.sql

@@ -141,29 +141,6 @@ $$;
 
 comment on function pgmq_public.read(queue_name text, sleep_seconds integer, n integer) is 'Reads up to "n" messages from the specified queue with an optional "sleep_seconds" (visibility timeout).';
 
--- Create receive function (alias for read with different parameter names for E2E test compatibility)
-create or replace function pgmq_public.receive(
-    queue_name text,
-    vt integer,
-    qty integer
-)
-  returns setof pgmq.message_record
-  language plpgsql
-  set search_path = ''
-as $$
-begin
-    return query
-    select *
-    from pgmq.read(
-        queue_name := queue_name,
-        vt := vt,
-        qty := qty
-    );
-end;
-$$;
-
-comment on function pgmq_public.receive(queue_name text, vt integer, qty integer) is 'Alias for read() - reads messages from the specified queue with visibility timeout.';
-
 -- Grant execute permissions on wrapper functions to roles
 grant execute on function pgmq_public.pop(text) to postgres, service_role, anon, authenticated;
 grant execute on function pgmq.pop(text) to postgres, service_role, anon, authenticated;
@@ -174,8 +151,6 @@ grant execute on function pgmq.send(text, jsonb, integer) to postgres, service_r
 grant execute on function pgmq_public.send_batch(text, jsonb[], integer) to postgres, service_role, anon, authenticated;
 grant execute on function pgmq.send_batch(text, jsonb[], integer) to postgres, service_role, anon, authenticated;
 
-grant execute on function pgmq_public.receive(text, integer, integer) to postgres, service_role, anon, authenticated;
-
 grant execute on function pgmq_public.archive(text, bigint[]) to postgres, service_role, anon, authenticated;
 
 grant execute on function pgmq_public.delete(text, bigint) to postgres, service_role, anon, authenticated;

dev-packages/e2e-tests/test-applications/supabase-nextjs/tests/performance.test.ts

@@ -62,8 +62,7 @@ test('Sends server-side Supabase RPC spans and breadcrumbs', async ({ page, base
   expect(rpcSpan).toBeDefined();
   expect(rpcSpan?.data).toEqual(
     expect.objectContaining({
-      'db.operation': 'insert',
-      'db.table': 'get_supabase_status',
+      'db.operation': 'rpc',
       'db.system': 'postgresql',
       'sentry.op': 'db',
       'sentry.origin': 'auto.db.supabase',
@@ -76,7 +75,7 @@ test('Sends server-side Supabase RPC spans and breadcrumbs', async ({ page, base
     transactionEvent.breadcrumbs?.some(
       breadcrumb =>
         breadcrumb?.type === 'supabase' &&
-        breadcrumb?.category === 'db.insert' &&
+        breadcrumb?.category === 'db.rpc' &&
         typeof breadcrumb?.message === 'string' &&
         breadcrumb.message.includes('get_supabase_status'),
     ),

packages/core/src/integrations/supabase.ts

@@ -901,7 +901,7 @@ function instrumentRpcConsumer(
     return instrumentGenericRpc(target, thisArg, argumentsList);
   }
 
-  const typedParams = queueParams as { queue_name?: string; vt?: number; qty?: number };
+  const typedParams = queueParams as { queue_name?: string; sleep_seconds?: number; n?: number };
   const queueName = typedParams.queue_name;
 
   if (!queueName) {
@@ -1019,7 +1019,7 @@ function createRpcProxyHandler(): ProxyHandler<(...args: unknown[]) => unknown>
       try {
         const normalizedName = normalizeRpcFunctionName(argumentsList[0]);
         const isProducerSpan = normalizedName === 'send' || normalizedName === 'send_batch';
-        const isConsumerSpan = normalizedName === 'pop' || normalizedName === 'receive' || normalizedName === 'read';
+        const isConsumerSpan = normalizedName === 'pop' || normalizedName === 'read';
 
         if (isProducerSpan) {
           return instrumentRpcProducer(target, thisArg, argumentsList);
@@ -1058,8 +1058,7 @@ function instrumentGenericRpc(
   builder.then = function (onfulfilled?: (value: unknown) => unknown, onrejected?: (reason: unknown) => unknown) {
     const attributes: Record<string, unknown> = {
       'db.system': 'postgresql',
-      'db.operation': 'insert', // RPC calls use POST which maps to 'insert'
-      'db.table': functionName,
+      'db.operation': 'rpc',
       [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.db.supabase',
       [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'db',
     };
@@ -1091,7 +1090,7 @@ function instrumentGenericRpc(
 
               const breadcrumb: SupabaseBreadcrumb = {
                 type: 'supabase',
-                category: 'db.insert',
+                category: 'db.rpc',
                 message: `rpc(${functionName})`,
               };