Supabase•8mo ago

Queues archiving via edge function

I'm working on implementing Queues (pgmq) like suggested in the tutorial. I have RLS and Expose Queues via PostgREST enabled but i'm only accessing them directly with edge functions. I can't seem to get the edge function to archive it. Are there any edge function examples out there that share the code? I don't see the edge fucntion code in the example https://youtu.be/UEwfaElBnZk . Everything appears to be working other than the deno code to move it to the archive a_queue_name. As in my function is adding, processing, and removing them from the queue... just not the history.

Supabase

YouTube

We added the missing piece! 🧩

Supabase Queues allow you to store a sequence of messages allowing for batch processing, or long running asynchronous tasks. In this video, Jon Meyers demonstrates how to add and remove messages in the Queue from a Supabase Edge Function, and how to retry a message if it fails to be processed. ✍️ Supabase Queues (blog): https://supabase.com/blo...

4 Replies

tomaspozo•8mo ago

I don't know of official samples with these implementations, however I can share how I have implemented on a project of my own... First I like to create a client for queues:

import { createClient } from "npm:@supabase/supabase-js@2";

export const supabaseQueue = (authHeader: string) =>
  createClient(
    Deno.env.get("SUPABASE_URL")!,
    Deno.env.get("SUPABASE_ANON_KEY")!,
    {
      global: {
        headers: { Authorization: authHeader },
      },
      db: {
        schema: "pgmq_public",
      },
    },
  );

import { createClient } from "npm:@supabase/supabase-js@2";

export const supabaseQueue = (authHeader: string) =>
  createClient(
    Deno.env.get("SUPABASE_URL")!,
    Deno.env.get("SUPABASE_ANON_KEY")!,
    {
      global: {
        headers: { Authorization: authHeader },
      },
      db: {
        schema: "pgmq_public",
      },
    },
  );

and later on the functions I imported it and use it like this:

// To read 10 items from the queue and process them on my function
const queueRequest = await supabaseQueueClient.rpc("read", {
      queue_name: "remove_deleted_photoshoots_photos",
      sleep_seconds: 60,
      n: 10,
    });

// To read 10 items from the queue and process them on my function
const queueRequest = await supabaseQueueClient.rpc("read", {
      queue_name: "remove_deleted_photoshoots_photos",
      sleep_seconds: 60,
      n: 10,
    });

// When I successfully process an item of the queue, I delete it with the message id
const deleteRequest = await supabaseQueueClient.rpc("delete", {
        queue_name: "remove_deleted_photoshoots_photos",
        message_id: item.msg_id,
      });

// When I successfully process an item of the queue, I delete it with the message id
const deleteRequest = await supabaseQueueClient.rpc("delete", {
        queue_name: "remove_deleted_photoshoots_photos",
        message_id: item.msg_id,
      });

And I call the edge function every X time with a cron.

SaltedWoundOP•8mo ago

Thank you for taking the time to share your code - it's very similar to what I'm doing. Does this move the message to the archive queue?
My pops and deletes work, but I'm not seeing the archive. Maybe my assumption is flawed, but I think I should be able to go to my pgmq schema and see both tables q_queuename and a_queuename tables. I do see them and can see the messages queued up in the q_ table but never see them in the a_ table.
I've tried many different forms of the following and keep getting similar errors:

Attempting to archive message:
Archive operation completed:  
{
  success: false,
  message_id: 120,
  queue: "queuename"
}

{
  queue_name: "queuename",
  message_id: 120,
  msg_details: {
    read_ct: 0,
    enqueued_at: "2025-01-27T18:47:12.449099+00:00",
    vt: "2025-01-27T18:47:12.450303+00:00"
  }
}

And here's the code I'm using for the archive operation:

const { data: archiveData, error: archiveError } = await queues.rpc('archive', { 
  queue_name: 'notification_queue',
  message_id: messageId
});

It seems with pgmq you need to have partitioning enabled to use the archive feature. I see Partitioned Queues are listed as 'coming soon' in the UI when you create the queue. I think that explains it.

tomaspozo•8mo ago

Any error in the postgres logs? If you attempt to archive from the SQL console, by running

SELECT pgmq.archive(
  queue_name => 'my_queue',
  msg_id     => 2
);

SELECT pgmq.archive(
  queue_name => 'my_queue',
  msg_id     => 2
);

does that work?

SaltedWoundOP•8mo ago

That works... interesting. 🙂

Gaming

Programming

Queues archiving via edge function

Did you find this page helpful?