n8n-nodes-supabase-queue-trigger

This is an n8n community node that lets you trigger workflows based on messages in Supabase Queues with support for both REST API and Edge Functions.

Supabase Queues is a Postgres-native durable Message Queue system with guaranteed delivery built on the pgmq database extension. This node provides polling-based triggers to monitor queue messages and start workflows automatically, with advanced operations powered by Supabase Edge Functions.

n8n is a fair-code licensed workflow automation platform.

Installation
Operations
Credentials
Edge Functions Setup
Compatibility
Usage
Resources

Installation

Follow the installation guide in the n8n community nodes documentation.

npm install n8n-nodes-supabase-queue-trigger

Operations

The Supabase Queue Trigger node supports three operations with two execution methods:

Available Operations

• Read (Peek): Read messages from the queue without removing them. Messages become temporarily invisible to other consumers during the visibility timeout period.
• Pop (Consume): Read and remove messages from the queue permanently. Requires Edge Function for true message consumption.
• Peek (No Side Effects): View messages without any modifications or side effects. Requires Edge Function.

Execution Methods

REST API Method (Default for Read)

• ✅ Fast setup - No additional configuration required
• ✅ Reliable - Uses Supabase's native REST API
• ⚠️ Limited - Only supports Read operation
• ⚠️ Read-only constraints - Cannot truly remove messages

Edge Function Method (Required for Pop/Peek)

• ✅ Full functionality - Supports all operations including real Pop
• ✅ True message consumption - Actually removes messages from queue
• ✅ No side effects peek - View messages without any changes
• ⚠️ Additional setup - Requires Edge Function deployment

Credentials

You need to configure Supabase Queue API credentials:

Prerequisites

1. A Supabase project with Queues enabled
2. Enable the pgmq extension in your database
3. For Pop/Peek operations: Deploy the provided Edge Function

Authentication Setup

1. Go to your Supabase project settings
2. Navigate to API settings
3. Copy your project URL and service role key
4. In n8n, create new "Supabase Queue API" credentials:
   • Supabase URL: Your project URL (e.g., https://your-project.supabase.co)
   • Service Role Key: Your service role secret key
   • Schema: public (default, where queue functions are created)

Queue Setup

Ensure your Supabase project has queues properly configured:

-- Enable the pgmq extension
CREATE EXTENSION IF NOT EXISTS pgmq;

-- Create a queue
SELECT pgmq.create('my-queue');

-- Add a test message
SELECT pgmq.send('my-queue', '{"type": "test", "data": "Hello World"}');

Edge Functions Setup

For Pop (Consume) and Peek operations, you need to deploy an Edge Function to your Supabase project.

1. Create the Edge Function

Create a new file in your Supabase project: supabase/functions/pgmq-queue/index.ts

import { serve } from "https://deno.land/std/http/server.ts";
import { createClient } from "https://esm.sh/@supabase/supabase-js@2";

interface QueueMessage {
  msg_id: number;
  read_ct: number;
  enqueued_at: string;
  vt: string;
  message: any;
}

serve(async (req) => {
  try {
    // CORS headers
    const corsHeaders = {
      "Access-Control-Allow-Origin": "*",
      "Access-Control-Allow-Headers":
        "authorization, x-client-info, apikey, content-type",
      "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
    };

    if (req.method === "OPTIONS") {
      return new Response(null, { headers: corsHeaders });
    }

    // Parse request
    const url = new URL(req.url);
    const operation = url.searchParams.get("operation") || "read";
    const queueName = url.searchParams.get("queue_name");
    const qty = parseInt(url.searchParams.get("qty") || "1");
    const vt = parseInt(url.searchParams.get("vt") || "30");

    if (!queueName) {
      return new Response(
        JSON.stringify({ error: "queue_name parameter is required" }),
        {
          status: 400,
          headers: { ...corsHeaders, "Content-Type": "application/json" },
        }
      );
    }

    // Setup Supabase client
    const supabaseUrl = Deno.env.get("SUPABASE_URL");
    const supabaseKey = Deno.env.get("SUPABASE_SERVICE_ROLE_KEY");
    const supabase = createClient(supabaseUrl, supabaseKey);

    let data: QueueMessage[] = [];
    let error: any = null;

    // Execute operation
    switch (operation) {
      case "read":
        const readResult = await supabase.rpc("sql", {
          query: `SELECT * FROM pgmq.read($1, $2, $3)`,
          params: [queueName, vt, qty],
        });
        data = readResult.data || [];
        error = readResult.error;
        break;

      case "pop":
        const popResults: QueueMessage[] = [];
        for (let i = 0; i < qty; i++) {
          const popResult = await supabase.rpc("sql", {
            query: `SELECT * FROM pgmq.pop($1)`,
            params: [queueName],
          });

          if (popResult.error) {
            error = popResult.error;
            break;
          }

          if (popResult.data && popResult.data.length > 0) {
            popResults.push(popResult.data[0]);
          } else {
            break;
          }
        }
        data = popResults;
        break;

      case "peek":
        const peekResult = await supabase.rpc("sql", {
          query: `
            SELECT msg_id, read_ct, enqueued_at, vt, message 
            FROM pgmq.q_${queueName} 
            WHERE vt <= now() 
            ORDER BY msg_id 
            LIMIT $1
          `,
          params: [qty],
        });
        data = peekResult.data || [];
        error = peekResult.error;
        break;
    }

    if (error) {
      return new Response(
        JSON.stringify({
          error: `Failed to ${operation} from queue`,
          details: error,
        }),
        {
          status: 500,
          headers: { ...corsHeaders, "Content-Type": "application/json" },
        }
      );
    }

    // Process message JSON
    const processedData = data.map((item) => {
      try {
        if (typeof item.message === "string") {
          item.message = JSON.parse(item.message);
        }
      } catch (parseError) {
        console.warn("Failed to parse message JSON:", parseError);
      }
      return item;
    });

    return new Response(JSON.stringify(processedData), {
      status: 200,
      headers: { ...corsHeaders, "Content-Type": "application/json" },
    });
  } catch (globalError) {
    console.error("Edge Function Error:", globalError);
    return new Response(
      JSON.stringify({
        error: "Internal server error",
        details: globalError.message,
      }),
      {
        status: 500,
        headers: {
          "Access-Control-Allow-Origin": "*",
          "Content-Type": "application/json",
        },
      }
    );
  }
});

2. Deploy the Edge Function

# Navigate to your Supabase project directory
cd your-supabase-project

# Deploy the function
supabase functions deploy pgmq-queue

# Verify deployment
supabase functions list

3. Test the Edge Function

# Test the deployed function
curl "https://your-project.supabase.co/functions/v1/pgmq-queue?operation=peek&queue_name=my-queue&qty=1" \
  -H "Authorization: Bearer YOUR_SERVICE_ROLE_KEY"

Compatibility

• Minimum n8n version: 0.198.0
• Supabase Postgres version: 15.6.1.143 or later (required for pgmq extension)
• Tested with n8n version: 1.0.0+
• Edge Functions: Requires Supabase Pro plan or higher for production use

Usage

Basic Setup

1. Add the "Supabase Queue Trigger" node to your workflow
2. Configure your Supabase Queue API credentials
3. Set the queue name you want to monitor
4. Choose the operation (Read, Pop, or Peek)
5. For Pop/Peek: Set the Edge Function name (default: pgmq-queue)
6. Configure polling settings
7. Activate the workflow

Configuration Options

Queue Name: The name of the Supabase queue to monitor (must match exactly)

Operation:

• Read (Peek): Uses REST API by default, messages remain in queue but become temporarily invisible
• Pop (Consume): Uses Edge Function, messages are permanently removed from queue
• Peek (No Side Effects): Uses Edge Function, view messages without any modifications

Messages Per Poll: Number of messages to fetch per poll (1-100)

Visibility Timeout: Time in seconds messages stay invisible after being read (Read operation only)

Edge Function Name: Name of the deployed Edge Function (required for Pop and Peek operations, default: pgmq-queue)

Advanced Options

Create Queue if Not Exists: Automatically create the queue if it doesn't exist

Use Edge Function for READ: Force READ operations to use Edge Function instead of REST API (can bypass some limitations)

Example Message Structure

Messages returned by the trigger have this structure:

{
  "id": 123,
  "enqueued_at": "2025-01-20T10:30:00Z",
  "vt": "2025-01-20T10:31:00Z",
  "read_ct": 1,
  "message": {
    "type": "email",
    "recipient": "[email protected]",
    "subject": "Welcome!"
  },
  "queue_name": "email-queue",
  "operation": "pop",
  "method": "edge_function",
  "raw": {
    /* complete original message object */
  }
}

Operation Comparison

| Feature | Read (REST API) | Read (Edge Function) | Pop (Edge Function) | Peek (Edge Function) | | --------------- | --------------------- | --------------------- | ---------------------- | -------------------- | | Setup Required | ✅ None | ⚠️ Edge Function | ⚠️ Edge Function | ⚠️ Edge Function | | Message Removal | ❌ No | ❌ No | ✅ Yes | ❌ No | | Side Effects | ⚠️ Visibility timeout | ⚠️ Visibility timeout | ✅ Removal | ❌ None | | Performance | ✅ Fast | ✅ Fast | ✅ Fast | ✅ Fast | | Use Case | Development, Testing | Advanced scenarios | Production consumption | Monitoring, Debug |

Sample Workflows

Basic Email Processing (Read)

Supabase Queue Trigger (email-queue, Read) → HTTP Request (send email) → Set Node (log result)

Order Processing with Consumption (Pop)

Supabase Queue Trigger (orders-queue, Pop) → Function Node (validate order) → HTTP Request (process payment) → Set Node (mark complete)

Queue Monitoring (Peek)

Supabase Queue Trigger (monitoring-queue, Peek) → Function Node (analyze metrics) → HTTP Request (send alert)

Troubleshooting

Common Issues

"Edge Function not found" error:

• Ensure the Edge Function is deployed: supabase functions list
• Verify the function name matches the node configuration
• Check if the function is accessible: test with curl

"Queue not found" error:

• Verify the queue name matches exactly
• Ensure the queue exists in your database: SELECT * FROM pgmq.list_queues()
• Check if pgmq extension is enabled: SELECT * FROM pg_extension WHERE extname = 'pgmq'

Authentication errors:

• Verify your Supabase URL and service role key
• Ensure the service role has necessary permissions
• Check if API is accessible from your n8n instance
• For Edge Functions: verify service role key has function execution permissions

No messages received:

• Verify there are messages in the queue: SELECT * FROM pgmq.q_your_queue_name
• Check polling intervals (messages might be in visibility timeout)
• For Pop operations: messages are consumed and won't appear again

Edge Function errors:

• Check Edge Function logs in Supabase Dashboard
• Verify environment variables are set correctly
• Test function directly before using in n8n

Performance Considerations

• Set appropriate polling intervals to balance responsiveness and database load
• Use Pop operation for high-throughput scenarios where message consumption is required
• Use Peek operation for monitoring without affecting queue state
• Use Read operation for development and testing
• Monitor your database and Edge Function performance
• Consider queue size and message processing rates

Migration from v1.0 to v2.0

If you're upgrading from a version that only supported Read operations:

1. No breaking changes - existing Read configurations continue to work
2. New operations available - Pop and Peek require Edge Function deployment
3. Enhanced functionality - Option to use Edge Functions for Read operations
4. Better error handling - More specific error messages for different scenarios

Resources

• Supabase Queues Documentation
• Supabase Edge Functions Documentation
• pgmq Extension Documentation
• n8n Community Nodes Documentation
• n8n Trigger Node Development

Contributing

Issues and feature requests are welcome! Please check the issues page before submitting.

Development Setup

# Clone the repository
git clone https://github.com/yourusername/n8n-nodes-supabase-queue-trigger
cd n8n-nodes-supabase-queue-trigger

# Install dependencies
npm install

# Build the node
npm run build

# Link for local development
npm link

License

MIT License - see LICENSE file for details.

Changelog

1.1.0

• ✅ Added Pop (Consume) operation with Edge Functions
• ✅ Added Peek (No Side Effects) operation
• ✅ Added option to use Edge Functions for Read operations
• ✅ Enhanced error handling and debugging
• ✅ Added method tracking in message output
• ✅ Improved documentation and setup guides

v1.0.0

• ✅ Initial release with Read (Peek) operation
• ✅ REST API integration
• ✅ Basic polling and message handling