Processing queue messages

Messages sitting in a queue need a worker to process them. On SupaHost the simplest worker is an Edge Function that drains the queue in batches, scheduled by pg_cron to run every minute (or more often). Nothing of yours has to stay running, and a failed message is retried automatically.

This page builds that worker for the receipts queue from Queues with pgmq: each message asks for a receipt email for a paid invoice.

How the pieces fit

1. Your app sends messages to the queue with pgmq_public.send from server code.
2. A cron job calls the worker function every minute with net.http_post.
3. The worker reads a batch with a visibility timeout, processes each message, and deletes the ones that succeeded.
4. Failed messages reappear after the timeout and are tried again on a later run. After too many attempts, the worker archives them so they stop blocking the queue.

Step 1: write the worker function

Create a function called receipt-worker in Dashboard → your project → Functions tab → New function, with this index.ts. Add a secret RESEND_API_KEY in the Secrets card (or swap in your own email provider). See Environment variables and secrets.

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

const QUEUE = 'receipts'
const BATCH_SIZE = 10
const VISIBILITY_SECONDS = 120  // longer than one batch can take
const MAX_ATTEMPTS = 5
const TIME_BUDGET_MS = 60_000   // stop reading new batches after a minute

const serviceKey = Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!
const admin = createClient(Deno.env.get('SUPABASE_URL')!, serviceKey, {
  auth: { persistSession: false },
})
const queue = admin.schema('pgmq_public')

type ReceiptMessage = {
  msg_id: number
  read_ct: number
  enqueued_at: string
  vt: string
  message: { invoiceId: number; email: string }
}

async function sendReceipt(invoiceId: number, email: string) {
  const res = await fetch('https://api.resend.com/emails', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${Deno.env.get('RESEND_API_KEY')}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      from: 'Billing <[email protected]>',
      to: email,
      subject: `Receipt for invoice ${invoiceId}`,
      text: `Thanks, we received your payment for invoice ${invoiceId}.`,
    }),
    signal: AbortSignal.timeout(15_000),
  })
  if (!res.ok) throw new Error(`Email API returned ${res.status}`)
}

Deno.serve(async (req) => {
  // Only the cron job (or you, testing as service_role) may run the worker
  if (req.headers.get('Authorization') !== `Bearer ${serviceKey}`) {
    return new Response('Forbidden', { status: 403 })
  }

  const started = Date.now()
  const stats = { sent: 0, failed: 0, gaveUp: 0 }

  while (Date.now() - started < TIME_BUDGET_MS) {
    const { data, error } = await queue.rpc('read', {
      queue_name: QUEUE,
      sleep_seconds: VISIBILITY_SECONDS,
      n: BATCH_SIZE,
    })
    if (error) return Response.json({ error: error.message }, { status: 500 })

    const messages = (data ?? []) as ReceiptMessage[]
    if (messages.length === 0) break // queue is empty

    for (const msg of messages) {
      if (msg.read_ct > MAX_ATTEMPTS) {
        // A poison message: keep it in the archive table and move on
        await queue.rpc('archive', { queue_name: QUEUE, message_id: msg.msg_id })
        console.error('Giving up on message', msg.msg_id, JSON.stringify(msg.message))
        stats.gaveUp++
        continue
      }

      try {
        await sendReceipt(msg.message.invoiceId, msg.message.email)
        await queue.rpc('delete', { queue_name: QUEUE, message_id: msg.msg_id })
        stats.sent++
      } catch (err) {
        // Not deleted, so it becomes visible again after VISIBILITY_SECONDS
        console.error('Message', msg.msg_id, 'failed:', String(err))
        stats.failed++
      }
    }
  }

  console.log('receipt-worker run', JSON.stringify(stats))
  return Response.json(stats)
})

Leave Require JWT on. The service_role key is a valid project JWT, so the cron job's request passes, and the explicit check at the top stops anyone calling the worker with the public anon key.

Why these numbers

• Time budget. A function run must start responding within 150 seconds. Stopping new batches after 60 seconds leaves room for the last batch to finish. See Edge Function limits.
• Visibility timeout. 120 seconds is longer than one batch of 10 can take with a 15-second timeout on each email call. If a message's processing could outlast the timeout, another run could pick it up at the same time.
• Memory. Each run gets 150 MB. Processing messages one at a time, as here, keeps memory flat. If you process a batch in parallel, keep the batch small.

Step 2: test it by hand

Send a test message from the Queues tab (open the receipts queue and send {"invoiceId": 1, "email": "[email protected]"}), then go to the Functions tab, use Test a function with POST as service_role, and call receipt-worker. The response shows the counts, and the Logs card on the same tab shows the console.log lines.

Step 3: schedule it with cron

Store the project URL and service_role key in Vault once (skip this if you did it for another job), then schedule the worker:

select vault.create_secret('https://app-one.acme.supahost.space', 'project_url');
select vault.create_secret('<your service_role key>', 'service_role_key');

select cron.schedule(
  'receipt-worker',
  '* * * * *',
  $$
  select net.http_post(
    url := (select decrypted_secret from vault.decrypted_secrets where name = 'project_url')
           || '/functions/v1/receipt-worker',
    headers := jsonb_build_object(
      'Authorization', 'Bearer ' || (select decrypted_secret from vault.decrypted_secrets where name = 'service_role_key')
    ),
    timeout_milliseconds := 90000
  );
  $$
);

The cron job itself finishes immediately: net.http_post queues the request and pg_net sends it in the background. The longer timeout_milliseconds lets pg_net wait for the worker's reply, so net._http_response records its result. See Scheduled jobs with pg_cron.

Keeping an eye on it

• Queues tab: Waiting should stay low and Oldest should stay recent. A growing backlog means the worker is failing or can't keep up.
• Functions tab → Logs: each run logs its counts, and failures log the message id and error. See Logs and debugging.
• Archived messages: anything the worker gave up on is in pgmq.a_receipts. Fix the cause, then send those messages again.

Scaling up

When one run a minute isn't enough:

• Run more often. Use a seconds schedule like '20 seconds'. Runs can overlap; that's safe, because a message read by one run is hidden from the others until its visibility timeout ends.
• Raise the batch size for quick messages, keeping the visibility timeout above the time a full batch can take.
• Split the work. Separate queues and workers for slow and fast jobs stop a backlog of one kind blocking the other.

Workers share the instance with your database and API, so watch overall load as you scale.

Other kinds of workers

• SQL-only work (for example rolling up rows) doesn't need a queue worker in a function at all. A cron job running the SQL directly is simpler.
• A long-running process on your own server can run the same read, process, delete loop with supabase-js and the service_role key, polling with a short sleep when the queue is empty.

Whatever runs the loop, the rules are the same: delete only after the work succeeds, make handlers safe to run twice, and set the visibility timeout longer than the work takes.

Related

• Queue messages reappearing (visibility timeout)
• Running functions on a schedule
• Example: sending email
• How do I run background jobs?