pgx

Documentation

Index

• func CreateTableSQL(tableName string) string
• type Config
• type Option
  • func WithMaxRetries(n int) Option
  • func WithRetryAfter(d time.Duration) Option
  • func WithTableName(s string) Option
• type OptionFunc
  • func (f OptionFunc) Apply(config *Config)
• type Store
  • func New(pool *pgxpool.Pool, options ...Option) *Store
  • func NewStdLib(db *sql.DB, options ...Option) *Store
  • func (s *Store) FetchPending(ctx context.Context, limit int) ([]outbox.Message, error)
  • func (s *Store) MarkFailed(ctx context.Context, id string, err error) error
  • func (s *Store) MarkProcessed(ctx context.Context, ids ...string) error
  • func (s *Store) Save(ctx context.Context, tx any, msgs ...outbox.Message) error

Functions

func CreateTableSQL

func CreateTableSQL(tableName string) string

CreateTableSQL returns the DDL for creating the outbox messages table.

Types

type Config

type Config struct {
	TableName  string
	MaxRetries int
	RetryAfter time.Duration
}

Config holds the configuration for the PostgreSQL outbox store.

type Option

type Option interface {
	Apply(*Config)
}

An Option configures a Store instance.

func WithMaxRetries

func WithMaxRetries(n int) Option

WithMaxRetries sets the maximum number of retries before a message is marked as failed.

func WithRetryAfter

func WithRetryAfter(d time.Duration) Option

WithRetryAfter sets the duration after which a failed message is retried.

func WithTableName

func WithTableName(s string) Option

WithTableName sets the outbox table name.

type OptionFunc

type OptionFunc func(*Config)

OptionFunc is a function that configures a Store config.

func (OptionFunc) Apply

func (f OptionFunc) Apply(config *Config)

Apply calls f(config).

type Store

type Store struct {
	// contains filtered or unexported fields
}

Store implements outbox.Store using PostgreSQL.

func New

func New(pool *pgxpool.Pool, options ...Option) *Store

New creates a new outbox store using pgxpool.

func NewStdLib

func NewStdLib(db *sql.DB, options ...Option) *Store

NewStdLib creates a new outbox store using database/sql.

func (*Store) FetchPending

func (s *Store) FetchPending(ctx context.Context, limit int) ([]outbox.Message, error)

FetchPending atomically claims up to limit pending messages for processing.

func (*Store) MarkFailed

func (s *Store) MarkFailed(ctx context.Context, id string, err error) error

MarkFailed increments retries and records the error.

func (*Store) MarkProcessed

func (s *Store) MarkProcessed(ctx context.Context, ids ...string) error

MarkProcessed marks messages as successfully processed.

func (*Store) Save

func (s *Store) Save(ctx context.Context, tx any, msgs ...outbox.Message) error

Save persists messages within the given transaction. tx can be pgx.Tx, *sql.Tx, or nil.