# Processing queue messages

> Build a pgmq queue worker on SupaHost. An Edge Function reads batches from pgmq_public, retries failures with visibility timeouts, sets aside poison messages, and runs every minute from cron.

Source: https://supahost.app/docs/cron-queues/queue-consumers · Updated 2026-11-11

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](/docs/cron-queues/queues): 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](/docs/functions/secrets).

```ts
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 <billing@your-domain.com>',
      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](/docs/functions/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": "you@example.com"}`), 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:

```sql
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](/docs/cron-queues/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](/docs/functions/logs).
- **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)](/kb/queue-messages-reappearing)
- [Running functions on a schedule](/docs/functions/scheduling)
- [Example: sending email](/docs/functions/sending-email)
- [How do I run background jobs?](/kb/background-jobs)