Match with OTEL semantics improve tests

Author: onurtemizkan

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

@@ -0,0 +1,85 @@
+import { NextApiRequest, NextApiResponse } from 'next';
+import { createClient } from '@supabase/supabase-js';
+import * as Sentry from '@sentry/nextjs';
+
+// These are the default development keys for a local Supabase instance
+const NEXT_PUBLIC_SUPABASE_URL = 'http://localhost:54321';
+const SUPABASE_SERVICE_ROLE_KEY =
+  'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImV4cCI6MTk4MzgxMjk5Nn0.EGIM96RAZx35lJzdJsyH-qQwv8Hdp7fsn3W0YpN81IU';
+
+const supabaseClient = createClient(NEXT_PUBLIC_SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY, {
+  db: {
+    schema: 'pgmq_public',
+  },
+});
+
+Sentry.instrumentSupabaseClient(supabaseClient);
+
+export default async function handler(req: NextApiRequest, res: NextApiResponse) {
+  // Step 1: Batch produce multiple messages
+  const { data: sendData, error: sendError } = await supabaseClient.rpc('send_batch', {
+    queue_name: 'batch-flow-queue',
+    messages: [
+      {
+        taskType: 'email',
+        recipient: '[email protected]',
+        subject: 'Welcome!',
+      },
+      {
+        taskType: 'email',
+        recipient: '[email protected]',
+        subject: 'Verification',
+      },
+      {
+        taskType: 'sms',
+        recipient: '+1234567890',
+        message: 'Your code is 123456',
+      },
+    ],
+  });
+
+  if (sendError) {
+    return res.status(500).json({ error: `Send batch failed: ${sendError.message}` });
+  }
+
+  // Step 2: Consume multiple messages from the queue
+  const { data: receiveData, error: receiveError } = await supabaseClient.rpc('receive', {
+    queue_name: 'batch-flow-queue',
+    vt: 30,
+    qty: 3,
+  });
+
+  if (receiveError) {
+    return res.status(500).json({ error: `Receive failed: ${receiveError.message}` });
+  }
+
+  // Step 3: Process all messages
+  const processedMessages = receiveData?.map((msg: any) => ({
+    messageId: msg.msg_id,
+    taskType: msg.message?.taskType,
+    processed: true,
+  }));
+
+  // Step 4: Archive all processed messages
+  const messageIds = receiveData?.map((msg: any) => msg.msg_id).filter(Boolean);
+  if (messageIds && messageIds.length > 0) {
+    const { error: archiveError } = await supabaseClient.rpc('archive', {
+      queue_name: 'batch-flow-queue',
+      msg_ids: messageIds,
+    });
+
+    if (archiveError) {
+      return res.status(500).json({ error: `Archive failed: ${archiveError.message}` });
+    }
+  }
+
+  return res.status(200).json({
+    success: true,
+    batchSize: 3,
+    produced: { messageIds: sendData },
+    consumed: {
+      count: receiveData?.length || 0,
+      messages: processedMessages,
+    },
+  });
+}

dev-packages/e2e-tests/test-applications/supabase-nextjs/pages/api/queue/concurrent-operations.ts

@@ -0,0 +1,111 @@
+import { NextApiRequest, NextApiResponse } from 'next';
+import { createClient } from '@supabase/supabase-js';
+import * as Sentry from '@sentry/nextjs';
+
+// These are the default development keys for a local Supabase instance
+const NEXT_PUBLIC_SUPABASE_URL = 'http://localhost:54321';
+const SUPABASE_SERVICE_ROLE_KEY =
+  'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImV4cCI6MTk4MzgxMjk5Nn0.EGIM96RAZx35lJzdJsyH-qQwv8Hdp7fsn3W0YpN81IU';
+
+const supabaseClient = createClient(NEXT_PUBLIC_SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY, {
+  db: {
+    schema: 'pgmq_public',
+  },
+});
+
+Sentry.instrumentSupabaseClient(supabaseClient);
+
+export default async function handler(req: NextApiRequest, res: NextApiResponse) {
+  // Test concurrent queue operations to multiple queues
+  // This validates that instrumentation handles parallel operations correctly
+
+  try {
+    // Produce messages to 3 different queues concurrently
+    const produceOperations = await Promise.all([
+      supabaseClient.rpc('send', {
+        queue_name: 'concurrent-queue-1',
+        message: { queueId: 1, task: 'process-images' },
+      }),
+      supabaseClient.rpc('send', {
+        queue_name: 'concurrent-queue-2',
+        message: { queueId: 2, task: 'send-emails' },
+      }),
+      supabaseClient.rpc('send', {
+        queue_name: 'concurrent-queue-3',
+        message: { queueId: 3, task: 'generate-reports' },
+      }),
+    ]);
+
+    // Check for errors
+    const produceErrors = produceOperations
+      .map((op, idx) => (op.error ? { queue: idx + 1, error: op.error.message } : null))
+      .filter(Boolean);
+
+    if (produceErrors.length > 0) {
+      return res.status(500).json({ error: 'Some produce operations failed', details: produceErrors });
+    }
+
+    // Consume from all queues concurrently
+    const consumeOperations = await Promise.all([
+      supabaseClient.rpc('receive', {
+        queue_name: 'concurrent-queue-1',
+        vt: 30,
+        qty: 1,
+      }),
+      supabaseClient.rpc('receive', {
+        queue_name: 'concurrent-queue-2',
+        vt: 30,
+        qty: 1,
+      }),
+      supabaseClient.rpc('receive', {
+        queue_name: 'concurrent-queue-3',
+        vt: 30,
+        qty: 1,
+      }),
+    ]);
+
+    // Process results
+    const consumeErrors = consumeOperations
+      .map((op, idx) => (op.error ? { queue: idx + 1, error: op.error.message } : null))
+      .filter(Boolean);
+
+    if (consumeErrors.length > 0) {
+      return res.status(500).json({ error: 'Some consume operations failed', details: consumeErrors });
+    }
+
+    // Archive all messages concurrently
+    const messageIds = consumeOperations.map((op, idx) => ({
+      queue: `concurrent-queue-${idx + 1}`,
+      msgId: op.data?.[0]?.msg_id,
+    }));
+
+    await Promise.all(
+      messageIds
+        .filter(m => m.msgId)
+        .map(m =>
+          supabaseClient.rpc('archive', {
+            queue_name: m.queue,
+            msg_ids: [m.msgId],
+          }),
+        ),
+    );
+
+    return res.status(200).json({
+      success: true,
+      concurrentOperations: {
+        queuesProcessed: 3,
+        produced: produceOperations.map(op => op.data),
+        consumed: consumeOperations.map((op, idx) => ({
+          queue: idx + 1,
+          messageId: op.data?.[0]?.msg_id,
+          task: op.data?.[0]?.message?.task,
+        })),
+      },
+    });
+  } catch (error) {
+    Sentry.captureException(error);
+    return res.status(500).json({
+      error: error instanceof Error ? error.message : 'Unknown error',
+    });
+  }
+}

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

@@ -0,0 +1,78 @@
+import { NextApiRequest, NextApiResponse } from 'next';
+import { createClient } from '@supabase/supabase-js';
+import * as Sentry from '@sentry/nextjs';
+
+// These are the default development keys for a local Supabase instance
+const NEXT_PUBLIC_SUPABASE_URL = 'http://localhost:54321';
+const SUPABASE_SERVICE_ROLE_KEY =
+  'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImV4cCI6MTk4MzgxMjk5Nn0.EGIM96RAZx35lJzdJsyH-qQwv8Hdp7fsn3W0YpN81IU';
+
+const supabaseClient = createClient(NEXT_PUBLIC_SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY, {
+  db: {
+    schema: 'pgmq_public',
+  },
+});
+
+Sentry.instrumentSupabaseClient(supabaseClient);
+
+export default async function handler(req: NextApiRequest, res: NextApiResponse) {
+  // Step 1: Produce a message that will cause processing error
+  const { data: sendData, error: sendError } = await supabaseClient.rpc('send', {
+    queue_name: 'error-flow-queue',
+    message: {
+      action: 'divide',
+      numerator: 100,
+      denominator: 0, // This will cause an error
+    },
+  });
+
+  if (sendError) {
+    return res.status(500).json({ error: `Send failed: ${sendError.message}` });
+  }
+
+  // Step 2: Consume the message
+  const { data: receiveData, error: receiveError } = await supabaseClient.rpc('receive', {
+    queue_name: 'error-flow-queue',
+    vt: 30,
+    qty: 1,
+  });
+
+  if (receiveError) {
+    return res.status(500).json({ error: `Receive failed: ${receiveError.message}` });
+  }
+
+  // Step 3: Process the message - this will throw an error
+  const message = receiveData?.[0];
+
+  try {
+    if (message?.message?.denominator === 0) {
+      throw new Error('Division by zero error in queue processor');
+    }
+
+    // Simulate successful processing (won't be reached in this flow)
+    const result = message.message.numerator / message.message.denominator;
+
+    return res.status(200).json({
+      success: true,
+      result,
+      messageId: message?.msg_id,
+    });
+  } catch (error) {
+    // Capture the error with Sentry
+    Sentry.captureException(error, scope => {
+      scope.setContext('queue', {
+        queueName: 'error-flow-queue',
+        messageId: message?.msg_id,
+        message: message?.message,
+      });
+      return scope;
+    });
+
+    // Return error response
+    return res.status(500).json({
+      success: false,
+      error: error instanceof Error ? error.message : 'Unknown error',
+      messageId: message?.msg_id,
+    });
+  }
+}

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

@@ -0,0 +1,67 @@
+import { NextApiRequest, NextApiResponse } from 'next';
+import { createClient } from '@supabase/supabase-js';
+import * as Sentry from '@sentry/nextjs';
+
+// These are the default development keys for a local Supabase instance
+const NEXT_PUBLIC_SUPABASE_URL = 'http://localhost:54321';
+const SUPABASE_SERVICE_ROLE_KEY =
+  'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImV4cCI6MTk4MzgxMjk5Nn0.EGIM96RAZx35lJzdJsyH-qQwv8Hdp7fsn3W0YpN81IU';
+
+const supabaseClient = createClient(NEXT_PUBLIC_SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY, {
+  db: {
+    schema: 'pgmq_public',
+  },
+});
+
+Sentry.instrumentSupabaseClient(supabaseClient);
+
+export default async function handler(req: NextApiRequest, res: NextApiResponse) {
+  // Step 1: Produce a message to the queue
+  const { data: sendData, error: sendError } = await supabaseClient.rpc('send', {
+    queue_name: 'e2e-flow-queue',
+    message: {
+      action: 'process_order',
+      orderId: 'ORDER-123',
+      timestamp: new Date().toISOString(),
+    },
+  });
+
+  if (sendError) {
+    return res.status(500).json({ error: `Send failed: ${sendError.message}` });
+  }
+
+  // Step 2: Consume the message from the queue (with VT=30 seconds)
+  const { data: receiveData, error: receiveError } = await supabaseClient.rpc('receive', {
+    queue_name: 'e2e-flow-queue',
+    vt: 30,
+    qty: 1,
+  });
+
+  if (receiveError) {
+    return res.status(500).json({ error: `Receive failed: ${receiveError.message}` });
+  }
+
+  // Step 3: Process the message (simulate business logic)
+  const processedMessage = receiveData?.[0];
+
+  // Step 4: Archive the message after successful processing
+  if (processedMessage?.msg_id) {
+    const { error: archiveError } = await supabaseClient.rpc('archive', {
+      queue_name: 'e2e-flow-queue',
+      msg_ids: [processedMessage.msg_id],
+    });
+
+    if (archiveError) {
+      return res.status(500).json({ error: `Archive failed: ${archiveError.message}` });
+    }
+  }
+
+  return res.status(200).json({
+    success: true,
+    produced: { messageId: sendData },
+    consumed: {
+      messageId: processedMessage?.msg_id,
+      message: processedMessage?.message,
+    },
+  });
+}

dev-packages/e2e-tests/test-applications/supabase-nextjs/pages/api/rpc/status.ts

@@ -0,0 +1,21 @@
+import type { NextApiRequest, NextApiResponse } from 'next';
+import { getSupabaseClient } from '@/lib/initSupabaseAdmin';
+
+const supabaseClient = getSupabaseClient();
+
+type Data = {
+  data: unknown;
+  error: unknown;
+};
+
+export default async function handler(req: NextApiRequest, res: NextApiResponse<Data>) {
+  const { data, error } = await supabaseClient.rpc('get_supabase_status');
+
+  if (error) {
+    console.warn('Supabase RPC status check failed', error);
+    res.status(500).json({ data, error });
+    return;
+  }
+
+  res.status(200).json({ data, error });
+}