### Install go-queue

Source: https://github.com/shyim/go-queue/blob/main/middleware/otel/README.md

Install the go-queue library using go get.

```sh
go get github.com/shyim/go-queue
```

--------------------------------

### Quick Start: go-queue Example

Source: https://github.com/shyim/go-queue/blob/main/README.md

Demonstrates setting up a message bus with AMQP transport, defining message types, registering handlers, dispatching a message, and running a worker. Ensure AMQP is running and accessible.

```go
package main

import (
	"context"
	"fmt"
	"log"
	"os/signal"
	"syscall"

	"github.com/shyim/go-queue"
	"github.com/shyim/go-queue/transport/amqp"
)

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

type ResizeImage struct {
	URL    string `json:"url"`
	Width  int    `json:"width"`
	Height int    `json:"height"`
}

func main() {

transport := amqp.NewTransport(amqp.Config{
	DSN:      "amqp://guest:guest@localhost:5672/",
	Exchange: "app",
	Queue:    "default",
})
defer transport.Close()

bus := queue.NewBus()
bus.AddTransport("async", transport)

queue.HandleFunc[SendEmail](bus, "async", func(ctx context.Context, msg SendEmail) error {
	fmt.Printf("Sending email to %s: %s\n", msg.To, msg.Subject)
	return nil
})

queue.HandleFunc[ResizeImage](bus, "async", func(ctx context.Context, msg ResizeImage) error {
	fmt.Printf("Resizing %s to %dx%d\n", msg.URL, msg.Width, msg.Height)
	return nil
})

if err := bus.Setup(context.Background()); err != nil {
	log.Fatal(err)
}

ctx := context.Background()
if err := queue.Dispatch(ctx, bus, SendEmail{
		To:      "user@example.com",
		Subject: "Welcome!",
		Body:    "Hello!",
	}); err != nil {
	log.Fatal(err)
}

ctx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
defer stop()

worker := queue.NewWorker(bus, queue.WorkerConfig{Concurrency: 4})
if err := worker.Run(ctx); err != nil && err != context.Canceled {
	log.Fatal(err)
}
}
```

--------------------------------

### Basic OpenTelemetry Setup

Source: https://github.com/shyim/go-queue/blob/main/middleware/otel/README.md

Register OpenTelemetry middleware for both dispatch (producer) and worker (consumer) to automatically trace and collect metrics. This setup ensures all queue operations are observed.

```go
import queueotel "github.com/shyim/go-queue/middleware/otel"

bus := queue.NewBus()
bus.AddTransport("async", transport)

// Producer tracing: creates spans and injects traceparent into headers.
bus.AddDispatchMiddleware(queueotel.DispatchMiddleware())

// Consumer tracing + metrics.
worker := queue.NewWorker(bus, queue.WorkerConfig{
	Middleware: []queue.Middleware{
		queueotel.Middleware(),        // tracing
		queueotel.MetricsMiddleware(), // metrics
	},
})

// Then use queue.Dispatch as usual — tracing is automatic.
queue.Dispatch(ctx, bus, MyMsg{...})
```

--------------------------------

### Implement Optional SetupTransport Interface

Source: https://github.com/shyim/go-queue/blob/main/README.md

Optionally implement the `SetupTransport` interface for automatic infrastructure setup when using a custom transport.

```go
type SetupTransport interface {
	Setup(ctx context.Context) error
}
```

--------------------------------

### Implement Logging Middleware

Source: https://github.com/shyim/go-queue/blob/main/README.md

Create a middleware function to log message processing start, duration, and completion status.

```go
func loggingMiddleware(ctx context.Context, env *queue.Envelope, next func(context.Context, *queue.Envelope) error) error {
	slog.Info("processing", "type", env.Type, "id", env.ID)
	start := time.Now()
	err := next(ctx, env)
	slog.Info("done", "type", env.Type, "duration", time.Since(start), "error", err)
	return err
}
```

--------------------------------

### Unit Test with Memory Transport

Source: https://github.com/shyim/go-queue/blob/main/README.md

Use the in-memory transport for unit tests. This setup involves creating a new bus, adding the memory transport, registering a handler, dispatching a message, and running a worker to process messages.

```go
import "github.com/shyim/go-queue/transport/memory"

bus := queue.NewBus()
transport := memory.NewTransport()
bus.AddTransport("async", transport)

queue.HandleFunc[MyMsg](bus, "async", handler)
queue.Dispatch(ctx, bus, MyMsg{...})

ctx, cancel := context.WithTimeout(ctx, time.Second)
def cancel()
worker := queue.NewWorker(bus, queue.WorkerConfig{})
worker.Run(ctx)

acked := transport.Acked()
nacked := transport.Nacked()
```

--------------------------------

### Basic AMQP Transport Usage

Source: https://github.com/shyim/go-queue/blob/main/transport/amqp/README.md

Initialize and set up the AMQP transport with a basic configuration. Ensure to close the transport when done.

```go
import "github.com/shyim/go-queue/transport/amqp"

transport := amqp.NewTransport(amqp.Config{
	DSN:      "amqp://guest:guest@localhost:5672/",
	Exchange: "myapp",
	Queue:    "tasks",
})
def transport.Close()

bus := queue.NewBus()
bus.AddTransport("async", transport)
bus.Setup(ctx)
```

--------------------------------

### Configure a Worker

Source: https://github.com/shyim/go-queue/blob/main/README.md

Set up a worker with custom concurrency, timeouts, retry strategies, and error/failure handlers.

```go
worker := queue.NewWorker(bus, queue.WorkerConfig{
	Concurrency:     8,
	HandlerTimeout:  30 * time.Second,
	ShutdownTimeout: 60 * time.Second, // default 30s

	RetryStrategy: queue.RetryStrategy{
		MaxRetries: 5,
		Delay:      1 * time.Second,
		Multiplier: 2.0,
		MaxDelay:   1 * time.Minute,
		Jitter:     0.1,
	},

	ErrorHandler: func(ctx context.Context, env *queue.Envelope, err error) {
		slog.Error("message failed", "error", err, "type", env.Type)
	},

	FailureHandler: func(ctx context.Context, env *queue.Envelope, err error) {
		// dead-letter, alert, etc.
	},

	Middleware: []queue.Middleware{myLoggingMiddleware},
})
```

--------------------------------

### Initialize PostgreSQL Transport

Source: https://github.com/shyim/go-queue/blob/main/transport/postgres/README.md

Create a new PostgreSQL transport instance with connection details and table name. Ensure to close the transport when done.

```go
import "github.com/shyim/go-queue/transport/postgres"

transport := postgres.NewTransport(postgres.Config{
	DSN:   "postgres://user:pass@localhost:5432/mydb",
	Table: "queue_messages",
})
def transport.Close()

bus := queue.NewBus()
bus.AddTransport("pg", transport)
bus.Setup(ctx) // creates table, index, notify trigger
```

--------------------------------

### AMQP Transport Configuration Options

Source: https://github.com/shyim/go-queue/blob/main/transport/amqp/README.md

Explore the various configuration options available for the AMQP transport, including DSN, exchange, queue, and more. Defaults are provided for most options.

```go
amqp.Config{
	DSN:               "amqp://guest:guest@localhost:5672/",
	Exchange:          "myapp",           // exchange name (default: "messages")
	ExchangeType:      "direct",          // direct, topic, fanout, headers (default: "direct")
	Queue:             "tasks",           // queue name (default: "default")
	RoutingKey:        "tasks",           // routing key (default: "default")
	PrefetchCount:     10,                // consumer prefetch (default: 10)
	Durable:           true,              // survive broker restart (default: true)
	AutoSetup:         true,              // auto-declare exchange/queue/binding (default: true)
	ReconnectDelay:    5 * time.Second,   // delay between reconnection attempts (default: 5s)
	PublisherConfirms: true,              // wait for broker ack on publish (default: true)
	DelayedExchange:   false,             // see Delayed Messages below
	Deduplication:     nil,               // see Deduplication below
}
```

--------------------------------

### Initialize Transport with Existing Pool

Source: https://github.com/shyim/go-queue/blob/main/transport/postgres/README.md

Create a PostgreSQL transport instance using an existing pgxpool. The caller is responsible for closing the pool.

```go
import "github.com/jackc/pgx/v5/pgxpool"

pool, _ := pgxpool.New(ctx, "postgres://...")
transport := postgres.NewTransportFromPool(pool, postgres.Config{
	Table: "queue_messages",
})
// Caller is responsible for closing the pool.
```

--------------------------------

### Integrate OpenTelemetry with Go-Queue

Source: https://github.com/shyim/go-queue/blob/main/README.md

Register OpenTelemetry middleware once to automatically trace all dispatch and consume calls. Use `queueotel.DispatchMiddleware` for dispatch and `queueotel.Middleware` and `queueotel.MetricsMiddleware` for workers.

```go
import queueotel "github.com/shyim/go-queue/middleware/otel"

bus.AddDispatchMiddleware(queueotel.DispatchMiddleware())

worker := queue.NewWorker(bus, queue.WorkerConfig{
	Middleware: []queue.Middleware{
		queueotel.Middleware(),        // tracing
		queueotel.MetricsMiddleware(), // metrics
	},
})
```

--------------------------------

### PostgreSQL Transport Configuration Options

Source: https://github.com/shyim/go-queue/blob/main/transport/postgres/README.md

Configure the PostgreSQL transport with options like DSN, table name, notification channel, and intervals for polling, reaper, and visibility timeout.

```go
postgres.Config{
	DSN:               "postgres://...",
	Table:             "queue_messages",   // table name (default: "queue_messages")
	Channel:           "queue_notify",     // LISTEN/NOTIFY channel (default: "queue_notify")
	PollInterval:      5 * time.Second,    // fallback poll interval (default: 5s)
	ReaperInterval:    1 * time.Minute,    // stale message reclaim interval (default: 1m)
	VisibilityTimeout: 5 * time.Minute,    // max processing time before reclaim (default: 5m)
}
```

--------------------------------

### Custom Tracer Configuration

Source: https://github.com/shyim/go-queue/blob/main/middleware/otel/README.md

Configure custom OpenTelemetry tracers for both producer dispatch middleware and consumer middleware. This allows for application-specific tracer naming and configuration.

```go
tracer := otel.Tracer("my-app")
bus.AddDispatchMiddleware(queueotel.DispatchMiddlewareWithTracer(tracer))

// Consumer side:
queueotel.MiddlewareWithTracer(tracer)
```

--------------------------------

### Implement Custom Transport Interface for Go-Queue

Source: https://github.com/shyim/go-queue/blob/main/README.md

Implement the `Transport` interface to create a new message broker transport. The `Receive` method should return a channel for consuming messages.

```go
type Transport interface {
	Send(ctx context.Context, envelope *queue.Envelope) error
	Receive(ctx context.Context) (<-chan *queue.Envelope, error)
	Ack(ctx context.Context, envelope *queue.Envelope) error
	Nack(ctx context.Context, envelope *queue.Envelope, requeue bool) error
	Retry(ctx context.Context, envelope *queue.Envelope) error
}
```

--------------------------------

### Custom Meter Configuration

Source: https://github.com/shyim/go-queue/blob/main/middleware/otel/README.md

Configure a custom OpenTelemetry meter for recording metrics. This allows for application-specific meter naming and configuration, ensuring metrics are properly categorized.

```go
meter := otel.Meter("my-app")
queueotel.MetricsMiddlewareWithMeter(meter)
```

--------------------------------

### Dispatch Messages with Options

Source: https://github.com/shyim/go-queue/blob/main/README.md

Dispatch messages with custom options like delay, max retries, specific queues, or custom headers.

```go
queue.Dispatch(ctx, bus, msg, queue.WithDelay(5*time.Minute))
```

```go
queue.Dispatch(ctx, bus, msg, queue.WithMaxRetries(10))
```

```go
queue.Dispatch(ctx, bus, msg, queue.WithQueue("high-priority"))
```

```go
queue.Dispatch(ctx, bus, msg, queue.WithHeader("x-tenant", "acme"))
```

--------------------------------

### AMQP Message Deduplication Configuration

Source: https://github.com/shyim/go-queue/blob/main/transport/amqp/README.md

Enable message deduplication by providing DeduplicationConfig. This requires LavinMQ or the rabbitmq-message-deduplication plugin.

```go
transport := amqp.NewTransport(amqp.Config{
	DSN: "amqp://localhost:5672/",
	Deduplication: &amqp.DeduplicationConfig{
		CacheSize: 10000,       // number of IDs to remember
		CacheTTL:  60000,       // milliseconds, 0 = unlimited
	},
})
```

--------------------------------

### PostgreSQL Table Schema

Source: https://github.com/shyim/go-queue/blob/main/transport/postgres/README.md

The SQL schema for the `queue_messages` table, including fields for message ID, type, body, headers, status, attempts, and timestamps.

```sql
CREATE TABLE IF NOT EXISTS queue_messages (
    id            BIGSERIAL PRIMARY KEY,
    envelope_id   TEXT        NOT NULL,
    type          TEXT        NOT NULL,
    body          BYTEA       NOT NULL,
    headers       JSONB       NOT NULL DEFAULT '{}',
    status        TEXT        NOT NULL DEFAULT 'pending',
    attempts      INT         NOT NULL DEFAULT 0,
    max_attempts  INT         NOT NULL DEFAULT 0,
    last_error    TEXT,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    available_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    started_at    TIMESTAMPTZ
);
```

--------------------------------

### Dispatching a Delayed Message

Source: https://github.com/shyim/go-queue/blob/main/transport/postgres/README.md

Dispatch a message with a specified delay using `queue.WithDelay`, which sets the `available_at` timestamp for the message.

```go
queue.Dispatch(ctx, bus, msg, queue.WithDelay(10*time.Minute))
```

--------------------------------

### Consuming Messages with SELECT FOR UPDATE SKIP LOCKED

Source: https://github.com/shyim/go-queue/blob/main/transport/postgres/README.md

SQL query to atomically claim a pending message for processing, using `FOR UPDATE SKIP LOCKED` to allow concurrent consumption by multiple workers.

```sql
WITH next AS (
    SELECT id FROM queue_messages
    WHERE status = 'pending' AND available_at <= NOW()
    ORDER BY created_at LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE queue_messages SET status = 'processing', started_at = NOW(), attempts = attempts + 1
FROM next WHERE queue_messages.id = next.id
RETURNING ...
```

--------------------------------

### AMQP Delayed Messages Configuration

Source: https://github.com/shyim/go-queue/blob/main/transport/amqp/README.md

Configure the AMQP transport to use delayed messages by setting DelayedExchange to true. This requires LavinMQ or the rabbitmq_delayed_message_exchange plugin.

```go
transport := amqp.NewTransport(amqp.Config{
	DSN:             "amqp://localhost:5672/",
	Exchange:        "delayed",
	ExchangeType:    "direct",
	DelayedExchange: true,
})

queue.Dispatch(ctx, bus, msg, queue.WithDelay(10*time.Second))
```

--------------------------------

### Check PostgreSQL Transport Connection

Source: https://github.com/shyim/go-queue/blob/main/transport/postgres/README.md

Check if the PostgreSQL transport has active connections using the `IsConnected` method.

```go
if transport.IsConnected() {
	// pool has active connections
}
```

--------------------------------

### Reclaiming Stale Messages

Source: https://github.com/shyim/go-queue/blob/main/transport/postgres/README.md

SQL query executed by the reaper to reclaim messages that have been in the 'processing' state longer than the `VisibilityTimeout`.

```sql
UPDATE queue_messages SET status = 'pending', available_at = NOW()
WHERE status = 'processing' AND started_at < NOW() - interval '5 minutes'
```

--------------------------------

### Span Name Normalization

Source: https://github.com/shyim/go-queue/blob/main/middleware/otel/README.md

Customize span names for messages to prevent overly verbose type names from cluttering trace data. This applies normalization to both producer and consumer spans.

```go
bus.AddDispatchMiddleware(queueotel.DispatchMiddleware(
	queueotel.WithSpanNameNormalizer(queueotel.DefaultSpanNameNormalizer),
))

worker := queue.NewWorker(bus, queue.WorkerConfig{
	Middleware: []queue.Middleware{
		queueotel.Middleware(queueotel.WithSpanNameNormalizer(queueotel.DefaultSpanNameNormalizer)),
	},
})
```

--------------------------------

### Access Envelope in Handler

Source: https://github.com/shyim/go-queue/blob/main/README.md

Retrieve the message envelope from the context within a message handler to access its ID, type, and headers.

```go
queue.HandleFunc[MyMsg](bus, "async", func(ctx context.Context, msg MyMsg) error {
	env := queue.EnvelopeFromContext(ctx)
	slog.Info("processing", "id", env.ID, "type", env.Type)
	return nil
})
```

--------------------------------

### Handle Unrecoverable Errors in Go-Queue

Source: https://github.com/shyim/go-queue/blob/main/README.md

Use `queue.Unrecoverable` to signal errors that should not be retried by the queue. This is useful for messages that are fundamentally invalid.

```go
queue.HandleFunc[MyMsg](bus, "async", func(ctx context.Context, msg MyMsg) error {
	if msg.Invalid() {
		return queue.Unrecoverable(fmt.Errorf("invalid message: %s", msg.ID))
	}
	return process(msg)
})
```

--------------------------------

### Check AMQP Transport Connection Status

Source: https://github.com/shyim/go-queue/blob/main/transport/amqp/README.md

Check if the AMQP transport is currently connected to the broker using the IsConnected method.

```go
if transport.IsConnected() {
	// ready
}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.