postgres

Documentation

Index

• Constants
• Variables
• func Batch(cb BatchFunc) error
• func DBWithDefaultContext(db *gorm.DB, fc func(db *gorm.DB) error) error
• func DefaultQueryCtx() (context.Context, context.CancelFunc)
• func DefaultQueryCtxWithParent(ctx context.Context) (context.Context, context.CancelFunc)
• func GetTxFromContext(ctx context.Context) *gorm.DB
• func GormTransaction(ctx context.Context, db *gorm.DB, fc func(tx *gorm.DB) error, ...) (err error)
• func GormTransactionWithDefaultContext(db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) error
• func GormTransactionWithoutContext(db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) (err error)
• func InjectTxIntoContext(ctx context.Context, tx *gorm.DB) context.Context
• func IsSerializationAnomaly(err error) bool
• func MustSQLDB(db *gorm.DB) *sql.DB
• func MustSQLTx(db *gorm.DB) *sql.Tx
• func NewConnection(uri string, dialect string, config Config) (db *sqlx.DB, gormDB *gorm.DB, err error)
• func NewEventBroadcaster(uri url.URL, minReconnectInterval time.Duration, ...) *eventBroadcaster
• func SetLogAllQueries(db *gorm.DB, enabled bool)
• func SqlTransaction(ctx context.Context, rdb *sql.DB, fc func(tx *sqlx.Tx) error, ...) (err error)
• func SqlxTransaction(ctx context.Context, q Queryer, fc func(tx *sqlx.Tx) error, ...) (err error)
• func SqlxTransactionWithDefaultCtx(q Queryer, fc func(tx *sqlx.Tx) error, txOpts ...sql.TxOptions) (err error)
• func TryUnwrapGormDB(db *gorm.DB) *sqlx.DB
• func TxFromContext(ctx context.Context, db *gorm.DB) *gorm.DB
• func UnwrapGormDB(db *gorm.DB) *sqlx.DB
• func WrapDbWithSqlx(rdb *sql.DB) *sqlx.DB
• type BatchFunc
• type Config
• type Event
• type EventBroadcaster
• type Lock
  • func NewLock(ctx context.Context, id int64, db *sql.DB) (Lock, error)
  • func (l *Lock) Lock(ctx context.Context) (bool, error)
  • func (l *Lock) Unlock(ctx context.Context) error
  • func (l *Lock) WaitAndLock(ctx context.Context) error
• type Locker
• type NullEventBroadcaster
  • func NewNullEventBroadcaster() *NullEventBroadcaster
  • func (*NullEventBroadcaster) Close() error
  • func (*NullEventBroadcaster) Healthy() error
  • func (*NullEventBroadcaster) Notify(channel string, payload string) error
  • func (*NullEventBroadcaster) NotifyInsideGormTx(tx *gorm.DB, channel string, payload string) error
  • func (*NullEventBroadcaster) Ready() error
  • func (*NullEventBroadcaster) Start() error
  • func (ne *NullEventBroadcaster) Subscribe(channel, payloadFilter string) (Subscription, error)
• type NullSubscription
  • func (ns *NullSubscription) ChannelName() string
  • func (ns *NullSubscription) Close()
  • func (ns *NullSubscription) Events() <-chan Event
  • func (ns *NullSubscription) InterestedIn(event Event) bool
  • func (ns *NullSubscription) Send(event Event)
• type Queryer
  • func UnwrapGorm(db *gorm.DB) Queryer
• type Subscription
• type TransactionManager
  • func NewGormTransactionManager(db *gorm.DB) TransactionManager
• type TransactionOption
  • func WithTxOptions(txopts sql.TxOptions) TransactionOption
  • func WithoutDeadline() TransactionOption
• type TxFn

Constants

const (
	ChannelJobCreated   = "insert_on_jobs"
	ChannelJobDeleted   = "delete_from_jobs"
	ChannelRunStarted   = "pipeline_run_started"
	ChannelRunCompleted = "pipeline_run_completed"

	// Postgres channel to listen for new eth_txes
	ChannelInsertOnEthTx = "insert_on_eth_txes"
)

const (
	// LockTimeout controls the max time we will wait for any kind of database lock.
	// It's good to set this to _something_ because waiting for locks forever is really bad.
	LockTimeout = 15 * time.Second
	// IdleInTxSessionTimeout controls the max time we leave a transaction open and idle.
	// It's good to set this to _something_ because leaving transactions open forever is really bad.
	IdleInTxSessionTimeout = 1 * time.Hour
)

NOTE: In an ideal world the timeouts below would be set to something sane in the postgres configuration by the user. Since we do not live in an ideal world, it is necessary to override them here.

They cannot easily be set at a session level due to how Go's connection pooling works.

const BatchSize uint = 1000

BatchSize is the default number of DB records to access in one batch

const DefaultQueryTimeout = 10 * time.Second

Variables

var (
	DefaultSqlTxOptions = sql.TxOptions{

		Isolation: sql.LevelReadCommitted,
	}
)

var (
	ErrNoDeadlineSet = errors.New("no deadline set")
)

Functions

func Batch

func Batch(cb BatchFunc) error

Batch is an iterator for batches of records

func DBWithDefaultContext

func DBWithDefaultContext(db *gorm.DB, fc func(db *gorm.DB) error) error

func DefaultQueryCtx

func DefaultQueryCtx() (context.Context, context.CancelFunc)

DefaultQueryCtx returns a context with a sensible sanity limit timeout for SQL queries

func DefaultQueryCtxWithParent

func DefaultQueryCtxWithParent(ctx context.Context) (context.Context, context.CancelFunc)

DefaultQueryCtxWithParent returns a context with a sensible sanity limit timeout for SQL queries with the given parent context

func GetTxFromContext

func GetTxFromContext(ctx context.Context) *gorm.DB

GetTxFromContext extracts the tx from the context. Returns nil if no tx exists.

func GormTransaction

func GormTransaction(ctx context.Context, db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) (err error)

DEPRECATED: Use the transaction manager instead.

func GormTransactionWithDefaultContext

func GormTransactionWithDefaultContext(db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) error

DEPRECATED: Use the transaction manager instead.

func GormTransactionWithoutContext

func GormTransactionWithoutContext(db *gorm.DB, fc func(tx *gorm.DB) error, txOptss ...sql.TxOptions) (err error)

WARNING: Only use for nested txes inside ORM methods where you expect db to already have a ctx with a deadline.

func InjectTxIntoContext

func InjectTxIntoContext(ctx context.Context, tx *gorm.DB) context.Context

InjectTxIntoContext injects the tx into the context

func IsSerializationAnomaly

func IsSerializationAnomaly(err error) bool

func MustSQLDB

func MustSQLDB(db *gorm.DB) *sql.DB

func MustSQLTx

func MustSQLTx(db *gorm.DB) *sql.Tx

MustSQLDB panics if there is an error getting the underlying SQL TX

func NewConnection

func NewConnection(uri string, dialect string, config Config) (db *sqlx.DB, gormDB *gorm.DB, err error)

func NewEventBroadcaster

func NewEventBroadcaster(uri url.URL, minReconnectInterval time.Duration, maxReconnectDuration time.Duration) *eventBroadcaster

func SetLogAllQueries

func SetLogAllQueries(db *gorm.DB, enabled bool)

func SqlTransaction

func SqlTransaction(ctx context.Context, rdb *sql.DB, fc func(tx *sqlx.Tx) error, txOpts ...sql.TxOptions) (err error)

func SqlxTransaction

func SqlxTransaction(ctx context.Context, q Queryer, fc func(tx *sqlx.Tx) error, txOpts ...sql.TxOptions) (err error)

func SqlxTransactionWithDefaultCtx

func SqlxTransactionWithDefaultCtx(q Queryer, fc func(tx *sqlx.Tx) error, txOpts ...sql.TxOptions) (err error)

func TryUnwrapGormDB

func TryUnwrapGormDB(db *gorm.DB) *sqlx.DB

func TxFromContext

func TxFromContext(ctx context.Context, db *gorm.DB) *gorm.DB

TxFromContext extracts the tx from the context. If no transaction value is provided in the context, it returns the gorm.DB.

func UnwrapGormDB

func UnwrapGormDB(db *gorm.DB) *sqlx.DB

func WrapDbWithSqlx

func WrapDbWithSqlx(rdb *sql.DB) *sqlx.DB

Types

type BatchFunc

type BatchFunc func(offset, limit uint) (count uint, err error)

BatchFunc is the function to execute on each batch of records, should return the count of records affected

type Config

type Config struct {
	LogSQLStatements bool
	MaxOpenConns     int
	MaxIdleConns     int
}

type Event

type Event struct {
	Channel string
	Payload string
}

type EventBroadcaster

type EventBroadcaster interface {
	service.Service
	Subscribe(channel, payloadFilter string) (Subscription, error)
	Notify(channel string, payload string) error
	NotifyInsideGormTx(tx *gorm.DB, channel string, payload string) error
}

EventBroadcaster opaquely manages a collection of Postgres event listeners and broadcasts events to subscribers (with an optional payload filter).

type Lock

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

Lock implements the Locker interface.

func NewLock

func NewLock(ctx context.Context, id int64, db *sql.DB) (Lock, error)

NewLock returns a Lock with *sql.Conn

func (*Lock) Lock

func (l *Lock) Lock(ctx context.Context) (bool, error)

Lock obtains exclusive session level advisory lock if available. It’s similar to WaitAndLock, except it will not wait for the lock to become available. It will either obtain the lock and return true, or return false if the lock cannot be acquired immediately.

func (*Lock) Unlock

func (l *Lock) Unlock(ctx context.Context) error

Unlock releases the lock and DB connection.

func (*Lock) WaitAndLock

func (l *Lock) WaitAndLock(ctx context.Context) error

WaitAndLock obtains exclusive session level advisory lock. If another session already holds a lock on the same resource identifier, this function will wait until the resource becomes available. Multiple lock requests stack, so that if the resource is locked three times it must then be unlocked three times.

type Locker

type Locker interface {
	Lock(ctx context.Context) (bool, error)
	WaitAndLock(ctx context.Context) error
	Unlock(ctx context.Context) error
}

Locker is an interface for postgresql advisory locks.

type NullEventBroadcaster

type NullEventBroadcaster struct {
	Sub *NullSubscription
}

NullEventBroadcaster implements null pattern for event broadcaster

func NewNullEventBroadcaster

func NewNullEventBroadcaster() *NullEventBroadcaster

func (*NullEventBroadcaster) Close

func (*NullEventBroadcaster) Close() error

func (*NullEventBroadcaster) Healthy

func (*NullEventBroadcaster) Healthy() error

func (*NullEventBroadcaster) Notify

func (*NullEventBroadcaster) Notify(channel string, payload string) error

func (*NullEventBroadcaster) NotifyInsideGormTx

func (*NullEventBroadcaster) NotifyInsideGormTx(tx *gorm.DB, channel string, payload string) error

func (*NullEventBroadcaster) Ready

func (*NullEventBroadcaster) Ready() error

func (*NullEventBroadcaster) Start

func (*NullEventBroadcaster) Start() error

func (*NullEventBroadcaster) Subscribe

func (ne *NullEventBroadcaster) Subscribe(channel, payloadFilter string) (Subscription, error)

type NullSubscription

type NullSubscription struct {
	Ch chan (Event)
}

func (*NullSubscription) ChannelName

func (ns *NullSubscription) ChannelName() string

func (*NullSubscription) Close

func (ns *NullSubscription) Close()

func (*NullSubscription) Events

func (ns *NullSubscription) Events() <-chan Event

func (*NullSubscription) InterestedIn

func (ns *NullSubscription) InterestedIn(event Event) bool

func (*NullSubscription) Send

func (ns *NullSubscription) Send(event Event)

type Queryer

type Queryer interface {
	sqlx.Ext
	sqlx.ExtContext
	QueryRow(query string, args ...interface{}) *sql.Row
}

func UnwrapGorm

func UnwrapGorm(db *gorm.DB) Queryer

type Subscription

type Subscription interface {
	Events() <-chan Event
	Close()

	ChannelName() string
	InterestedIn(event Event) bool
	Send(event Event)
}

Subscription represents a subscription to a Postgres event channel

type TransactionManager

type TransactionManager interface {
	Transact(TxFn, ...TransactionOption) error
	TransactWithContext(ctx context.Context, fn TxFn, optsFn ...TransactionOption) (err error)
}

func NewGormTransactionManager

func NewGormTransactionManager(db *gorm.DB) TransactionManager

type TransactionOption

type TransactionOption func(opts *transactionOptions)

TransactionManagerOption configures how we set up the transaction

func WithTxOptions

func WithTxOptions(txopts sql.TxOptions) TransactionOption

WithTxOptions returns a TransactionOption which sets the sql.TxOptions on the transaction.

func WithoutDeadline

func WithoutDeadline() TransactionOption

type TxFn

type TxFn func(context.Context) error

A TxFn is a function that will be called with a context which has a transaction injected as a value. This can be used for executing statements and queries against a database.