postgres

README

Migrations

Migrations are implemented with https://github.com/golang-migrate/migrate

Install the CLI

To add new migrations, install the CLI. The version of the CLI is not particularly important.

go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@v4.16.2

Adding migrations

Adding new migrations is done with the migrate create command

migrate create -dir backends/postgres/migrations -ext sql <descriptive_migration_name>

Running migrations

Migrations are run every time neoq initializes the postgres backend. There is no need to run migrations explicitly.

Documentation

Index

• Constants
• Variables
• func Backend(ctx context.Context, opts ...config.Option) (pb types.Backend, err error)
• func WithConnectionString(connectionString string) config.Option
• func WithTransactionTimeout(txTimeout int) config.Option
• type PgBackend
  • func (p *PgBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID string, err error)
  • func (p *PgBackend) SetLogger(logger logging.Logger)
  • func (p *PgBackend) Shutdown(ctx context.Context)
  • func (p *PgBackend) Start(ctx context.Context, queue string, h handler.Handler) (err error)
  • func (p *PgBackend) StartCron(ctx context.Context, cronSpec string, h handler.Handler) (err error)

Constants

const (
	DefaultPgConnectionString = "postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable"
	PendingJobIDQuery         = `` /* 157-byte string literal not displayed */

	PendingJobQuery = `` /* 250-byte string literal not displayed */

	FutureJobQuery = `` /* 196-byte string literal not displayed */

)

Variables

var (
	ErrCnxString              = errors.New("invalid connecton string: see documentation for valid connection strings")
	ErrDuplicateJob           = errors.New("duplicate job")
	ErrNoTransactionInContext = errors.New("context does not have a Tx set")
)

Functions

func Backend

func Backend(ctx context.Context, opts ...config.Option) (pb types.Backend, err error)

Backend initializes a new postgres-backed neoq backend

If the database does not yet exist, Neoq will attempt to create the database and related tables by default.

Backend requires that one of the config.ConfigOption is WithConnectionString

Connection strings may be a URL or DSN-style connection strings. The connection string supports multiple options detailed below.

options:

• pool_max_conns: integer greater than 0
• pool_min_conns: integer 0 or greater
• pool_max_conn_lifetime: duration string
• pool_max_conn_idle_time: duration string
• pool_health_check_period: duration string
• pool_max_conn_lifetime_jitter: duration string

Example DSN

user=worker password=secret host=workerdb.example.com port=5432 dbname=mydb sslmode=verify-ca pool_max_conns=10

Example URL

postgres://worker:secret@workerdb.example.com:5432/mydb?sslmode=verify-ca&pool_max_conns=10

func WithConnectionString

func WithConnectionString(connectionString string) config.Option

WithConnectionString configures neoq postgres backend to use the specified connection string when connecting to a backend

func WithTransactionTimeout

func WithTransactionTimeout(txTimeout int) config.Option

WithTransactionTimeout sets the time that PgBackend's transactions may be idle before its underlying connection is closed The timeout is the number of milliseconds that a transaction may sit idle before postgres terminates the transaction's underlying connection. The timeout should be longer than your longest job takes to complete. If set too short, job state will become unpredictable, e.g. retry counts may become incorrect.

Types

type PgBackend

type PgBackend struct {
	types.Backend
	// contains filtered or unexported fields
}

PgBackend is a Postgres-based Neoq backend

func (*PgBackend) Enqueue

func (p *PgBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID string, err error)

Enqueue adds jobs to the specified queue

func (*PgBackend) SetLogger

func (p *PgBackend) SetLogger(logger logging.Logger)

SetLogger sets this backend's logger

func (*PgBackend) Shutdown

func (p *PgBackend) Shutdown(ctx context.Context)

Shutdown shuts this backend down

func (*PgBackend) Start

func (p *PgBackend) Start(ctx context.Context, queue string, h handler.Handler) (err error)

Start starts processing jobs with the specified queue and handler

func (*PgBackend) StartCron

func (p *PgBackend) StartCron(ctx context.Context, cronSpec string, h handler.Handler) (err error)

StartCron starts processing jobs with the specified cron schedule and handler

See: https://pkg.go.dev/github.com/robfig/cron?#hdr-CRON_Expression_Format for details on the cron spec format