goqite Documentation

goqite is a persistent message queue library that uses a single database table to manage messages.

Key concepts include:

• Persistence: Messages are stored in a database (SQLite or PostgreSQL).
• Visibility Timeout: When a message is received, it is marked as received and becomes unavailable to other consumers until a timeout occurs. This prevents redelivery during active processing.
• Timeout Extension: You can extend the visibility timeout for long-running tasks to prevent the message from being redelivered while still being processed.
• Multiple Queues: You can manage multiple logical queues within a single database table by specifying a Name during initialization.
• Job Runner: A higher-level abstraction built on top of the queue to automate background task execution.

Install the goqite library using go get:

go get maragu.dev/goqite

To use PostgreSQL instead of the default SQLite, import the pgx driver and set the SQLFlavor to goqite.SQLFlavorPostgreSQL in goqite.NewOpts.

import _ "github.com/jackc/pgx/v5/stdlib"

q := goqite.New(goqite.NewOpts{
	DB:        db, // *sql.DB connected to PostgreSQL
	Name:      "jobs",
	SQLFlavor: goqite.SQLFlavorPostgreSQL,
})

The project provides a docker-compose.yml file to spin up a PostgreSQL 17 instance for testing purposes. This service is configured with the following credentials and settings:

• Image: postgres:17
• User: test
• Password: test
• Database: template1
• Host Port: 5433 (mapped to container port 5432)

You can start this environment using the command:

docker-compose up -d

services:
  postgres-test:
    image: postgres:17
    environment:
      POSTGRES_USER: "test"
      POSTGRES_PASSWORD: "test"
      POSTGRES_DB: "template1"
    ports:
      - "5433:5432"

Use goqite.New to create a new queue instance. The queue is backed by a SQL table and requires a *sql.DB connection and a unique Name.

Default values if not provided in NewOpts:

• MaxReceive: 3 (The maximum number of times a message can be received before it is ignored).
• Timeout: 5 seconds (The duration after which a message becomes available for re-receiving).
• SQLFlavor: Must be explicitly set to SQLFlavorSQLite or SQLFlavorPostgreSQL.

import (
	"database/sql"
	"time"
	"maragudk/maragudk/goqite"
)

opts := goqite.NewOpts{
	DB:        db, // your *sql.DB instance
	Name:      "my-queue",
	SQLFlavor: goqite.SQLFlavorSQLite,
	MaxReceive: 5,
	Timeout:    10 * time.Second,
}

queue := goqite.New(opts)

To use the core queue functionality, initialize a queue using goqite.New and use the following methods:

• Send(ctx, message): Sends a goqite.Message (containing a Body []byte) to the queue. You can also set a message delay.
• Receive(ctx): Retrieves a message from the queue. The message will not be available to others until its timeout expires.
• Extend(ctx, id, duration): Extends the visibility timeout for a specific message ID.
• Delete(ctx, id): Deletes a message from the queue so it is not redelivered.

// Initialize
q := goqite.New(goqite.NewOpts{
	DB:   db,
	Name: "jobs",
})

// Send
err = q.Send(ctx, goqite.Message{
	Body: []byte("payload"),
})

// Receive
m, err := q.Receive(ctx)

// Extend timeout
err = q.Extend(ctx, m.ID, time.Second)

// Delete
err = q.Delete(ctx, m.ID)

You must execute the appropriate schema in your database before using goqite. Use the SQLite schema for SQLite databases and the PostgreSQL schema for PostgreSQL databases.

-- SQLite Schema
create table goqite (
  id text primary key default ('m_' || lower(hex(randomblob(16))))
  -- ... (see full README for complete schema)
);

-- PostgreSQL Schema
create extension if not exists pgcrypto;

create table goqite (
  id text primary key default ('m_' || encode(gen_random_bytes(16), 'hex'))
  -- ... (see full README for complete schema)
);

Once a message is received, you can manage its lifecycle using its ID:

• Delete: Removes the message from the queue permanently. Use this after successfully processing a message.
• Extend: Increases the message's timeout by a specific delay from the current time. This is useful for long-running tasks to prevent the message from being re-delivered to another worker while still in progress.

// After processing
err := queue.Delete(ctx, msg.ID)

// If processing is taking longer than expected
err := queue.Extend(ctx, msg.ID, 30 * time.Second)

Use Receive to fetch the next available message. A message is available if its timeout has passed and its received count is less than the queue's MaxReceive setting.

• Receive returns nil, nil if no messages are available.
• ReceiveAndWait polls the queue at a specified interval until a message is found or the context is cancelled.

// Single attempt
msg, err := queue.Receive(ctx)
if err != nil {
	// handle error
}
if msg != nil {
	fmt.Printf("Received: %s\n", string(msg.Body))
}

// Polling until available
msg, err := queue.ReceiveAndWait(ctx, 500 * time.Millisecond)

Messages can be sent using Send (which uses an internal transaction) or SendTx (if you are already managing a transaction).

Message Fields:

• ID: The unique identifier (returned when using SendAndGetID).
• Body: The byte slice payload.
• Delay: A time.Duration representing how long to wait before the message becomes available for receipt.
• Priority: An integer where higher values are received first.

err := queue.Send(ctx, goqite.Message{
	Body:     []byte("hello world"),
	Priority: 10,
	Delay:    1 * time.Minute,
})

The Message struct defines the payload and metadata for items in the queue.

type Message struct {
	ID       ID
	Body     []byte
	Delay    time.Duration
	Priority int // Higher priority messages are received first
}