pgmq

A type-safe Go client for the pgmq PostgreSQL extension — a lightweight message queue on Postgres, so you don't need a separate queueing service.

Features

• Type-safe queues — Queue[T] marshals your payload type to and from JSON automatically
• Consumers — queue.Consume(ctx, handler) runs a worker loop with concurrency, automatic acknowledgment, retries via visibility timeouts, and poison-message protection
• Send options — delayed delivery by duration or absolute time, and message headers
• Batch operations — send, read, pop, archive, and delete in single round trips
• Long polling — pgmq.WithPoll() waits for messages instead of returning immediately
• Transactions — WithTx enqueues and dequeues atomically with your own database writes
• Queue management — standard, unlogged, and partitioned queues; purge, drop, archive, and metrics

Requirements

• PostgreSQL with the pgmq extension version 1.7 or newer
• Go 1.25+ and pgx/v5

Installation

go get github.com/joeychilson/pgmq

Usage

package main

import (
	"context"
	"log"
	"os"

	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/joeychilson/pgmq"
)

type EmailNotification struct {
	To      string `json:"to"`
	Subject string `json:"subject"`
	Body    string `json:"body"`
}

func main() {
	ctx := context.Background()

	pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		log.Fatalf("failed to connect to database: %v", err)
	}
	defer pool.Close()

	// Install the pgmq extension (once per database, requires the privilege
	// to create extensions) and create the queue if it doesn't exist.
	if err := pgmq.CreateExtension(ctx, pool); err != nil {
		log.Fatalf("failed to create extension: %v", err)
	}

	queue, err := pgmq.New[EmailNotification](pool, "email_notifications")
	if err != nil {
		log.Fatalf("failed to create queue instance: %v", err)
	}
	if err := queue.Create(ctx); err != nil {
		log.Fatalf("failed to create queue: %v", err)
	}

	// Produce a message.
	msgID, err := queue.Send(ctx, EmailNotification{
		To:      "user@example.com",
		Subject: "Welcome!",
		Body:    "Thanks for signing up.",
	})
	if err != nil {
		log.Fatalf("failed to send message: %v", err)
	}
	log.Printf("queued message %d", msgID)

	// Consume messages until ctx is canceled. Handled messages are archived;
	// failed messages reappear after the visibility timeout and are retried.
	err = queue.Consume(ctx, func(ctx context.Context, msg *pgmq.Message[EmailNotification]) error {
		log.Printf("sending email to %s: %s", msg.Message.To, msg.Message.Subject)
		return nil
	})
	if err != nil {
		log.Fatalf("consumer stopped: %v", err)
	}
}

Consumers

For control over concurrency, acknowledgment, and poison messages, configure a Consumer directly:

consumer := &pgmq.Consumer[EmailNotification]{
	Queue:        queue,
	Handler:      sendEmail,
	Concurrency:  4,               // handle up to 4 messages in parallel
	Ack:          pgmq.AckArchive, // or pgmq.AckDelete
	MaxReadCount: 5,               // archive poison messages after 5 attempts
	OnExhausted: func(ctx context.Context, msg *pgmq.Message[EmailNotification]) {
		log.Printf("giving up on message %d", msg.ID)
	},
	OnError: func(ctx context.Context, msg *pgmq.Message[EmailNotification], err error) {
		log.Printf("consumer error: %v", err)
	},
}
err := consumer.Run(ctx) // blocks until ctx is canceled

Sending

Send and SendBatch accept options for delayed delivery and message headers:

msgID, err := queue.Send(ctx, email,
	pgmq.WithDelay(10*time.Second),                       // or WithDelayUntil(time.Time)
	pgmq.WithHeaders(pgmq.Headers{"trace_id": "abc123"}), // read back via msg.Headers
)

Reading

Reads hide messages from other consumers for a visibility timeout (30s by default) instead of removing them — delete or archive a message to acknowledge it, or use Pop to read-and-delete in one step:

msg, err := queue.Read(ctx)                  // returns a nil message if the queue is empty
msg, err = queue.Read(ctx, pgmq.WithPoll())  // waits up to 5s for a message to arrive
msgs, err := queue.ReadBatch(ctx, 10, pgmq.WithVisibilityTimeout(time.Minute))

Transactions

Use WithTx to make queue operations part of your own transaction — a sent message only becomes visible if the transaction commits:

tx, err := pool.Begin(ctx)
if err != nil {
	return err
}
defer tx.Rollback(ctx)

if err := createOrder(ctx, tx, order); err != nil {
	return err
}
if _, err := queue.WithTx(tx).Send(ctx, OrderPlaced{OrderID: order.ID}); err != nil {
	return err
}
return tx.Commit(ctx)

Errors

Read and Pop return a nil message and nil error when no message is available. Operations on specific IDs return errors that can be matched with errors.Is:

if err := queue.Delete(ctx, msgID); errors.Is(err, pgmq.ErrMessageNotFound) {
	// already gone
}

Testing

The integration tests run against any PostgreSQL database with pgmq available and skip themselves if none is reachable:

docker run -d --name pgmq -e POSTGRES_PASSWORD=postgres -p 5433:5432 ghcr.io/pgmq/pg17-pgmq:latest
PGMQ_TEST_DATABASE_URL=postgres://postgres:postgres@localhost:5433/postgres go test ./...

License

MIT