Queues with pgmq

Queues let one part of your app hand work to another without doing it right away. A web request puts a message on a queue ("send the receipt for invoice 1042") and returns. A worker picks the message up later, does the work, and deletes it. If the worker fails, the message comes back and is tried again.

SupaHost Queues use pgmq, a Postgres extension. Queues are tables in your project's database, so messages are stored as safely as the rest of your data and are included in database backups.

Enable Queues

Queues are off until you turn them on for a project:

1. Go to Dashboard → your project → Queues tab.
2. Click Enable Queues (owner only).

This installs pgmq in the project's database and creates wrapper functions in the pgmq_public schema, which the Data API exposes:

The functions run with the caller's privileges (security invoker). Out of the box, execute on them is granted to service_role only. Signed-in users (authenticated) can see the schema but can't call the functions until you grant access yourself, so by default queues are only usable from server code.

Create a queue

On the Queues tab, click Create queue and enter a Queue name: 1 to 40 characters, lowercase letters, numbers and _. Tick Unlogged (faster, not crash-safe) only for messages you can afford to lose, since an unlogged queue table isn't written to Postgres's write-ahead log: it's emptied after a crash and its messages aren't in backups.

Each queue gets a table pgmq.q_<name> for live messages and pgmq.a_<name> for archived ones.

The Queues tab shows each queue with Waiting, Total sent and Oldest. Per queue you can open Messages (the newest 25), send a JSON message for testing, Purge all messages, or Delete the queue (you type its name to confirm).

Send messages from your server

This example is an invoicing app that queues a receipt email whenever a payment is recorded. It runs in a server route with the service_role key.

import { createClient } from '@supabase/supabase-js'

const admin = createClient(
  process.env.SUPABASE_URL!,               // https://app-one.acme.supahost.space
  process.env.SUPABASE_SERVICE_ROLE_KEY!,  // server only
  { auth: { persistSession: false } }
)

export async function queueReceipt(invoiceId: number, email: string) {
  const { data, error } = await admin.schema('pgmq_public').rpc('send', {
    queue_name: 'receipts',
    message: { invoiceId, email },
  })
  if (error) throw error
  return data // the new message id(s)
}

// Several at once, delayed by 60 seconds
await admin.schema('pgmq_public').rpc('send_batch', {
  queue_name: 'receipts',
  messages: [{ invoiceId: 1042, email: '[email protected]' }, { invoiceId: 1043, email: '[email protected]' }],
  sleep_seconds: 60,
})

Read and acknowledge messages

A consumer reads a batch, processes each message, and then deletes or archives it:

const { data: messages, error } = await admin.schema('pgmq_public').rpc('read', {
  queue_name: 'receipts',
  sleep_seconds: 60, // hide these messages from other readers for 60 s
  n: 10,             // up to 10 at a time
})
if (error) throw error

for (const msg of messages ?? []) {
  await sendReceiptEmail(msg.message.invoiceId, msg.message.email)

  await admin.schema('pgmq_public').rpc('delete', {
    queue_name: 'receipts',
    message_id: msg.msg_id,
  })
}

Each message returned by read and pop has msg_id, read_ct (how many times it has been read), enqueued_at, vt (when it becomes visible again) and message (your JSON).

archive instead of delete keeps a copy in pgmq.a_<name>, which is useful for auditing what was processed. Archived messages stay until you remove them, and they take up database space.

For a complete worker that runs on a schedule, see Processing queue messages.

Visibility timeouts

The sleep_seconds argument to read is the visibility timeout. While it's running, no other reader can get that message. Then:

• If your worker deletes or archives the message in time, it's done.
• If the worker crashes, times out or forgets to acknowledge it, the message becomes visible again when the timeout ends, and the next read returns it with read_ct increased by one.

That's what makes queues reliable: a message is only gone once you say so. It also means a message can be processed more than once, so make your handlers safe to repeat (for example, record "receipt sent for invoice 1042" and skip if it's already there).

Pick a timeout comfortably longer than one message takes to process. Too short and messages reappear while still being worked on, so two workers handle the same one. See Queue messages reappearing.

pop skips all this: it returns a message and deletes it in one step. If the worker then fails, the message is lost. Use it only for work that doesn't matter if it's skipped.

Using queues from the browser or app

Most apps keep queue access on the server. If you do want signed-in users to send messages directly (for example a feedback form that enqueues a job), you need to grant it yourself, as the Queues tab explains: grant execute on the pgmq_public functions you need to authenticated, and give authenticated access to the queue's table pgmq.q_<name> with RLS policies that say who may do what. Upstream Supabase documents the table privileges each function needs (send needs select and insert; read and pop need select and update; archive and delete need select and delete). Grant only what the client needs, which is usually send alone, and test it with a non-admin user.

When to use a queue

Related

• Processing queue messages
• Can I use Postgres as a message queue?
• How do I run background jobs?
• Queues API reference in Supabase's docs