pgsql

Documentation

Overview

AdvisoryLock implements primitives to use PostgreSQL's advisory locks.

These locks can be used to pipeline concurrent access between multiple database sessions. Keep in mind, both [Lock] and [TryLock] are stackable, can be called multiple times on the same session, requiring the same amount of calls to [Unlock] to free up the lock:

   var l := NewAdvisoryLock(conn, 32)
   l.Lock(context.Background()) // lock does not exist previously in this session, so locks successfully
   l.Lock(context.Background()) // lock already exists, but this increments the lock

   l.Unlock() // lock is not freed yet, as it was locked twice
   l.Unlock() // only here lock is released

See https://www.postgresql.org/docs/current/functions-admin.html#FUNCTIONS-ADVISORY-LOCKS for more details on the specifics

of PostgreSQL advisory locks

Note: since pgxpool is used, the lock will keep a connection from the connection pool for himself; this connection is freed when the number of locked unlocks is achieved

Index

• Constants
• func DialectOptions() *goqu.SQLDialectOptions
• func ForeignTableExists(ctx context.Context, db *sqlx.DB, tableName string, schema string) (bool, error)
• func GetServerVersion(db *sqlx.DB, ctx context.Context) (string, error)
• func NewClient(config *ClientConfig) (*db.SqlClient, error)
• func NewMigrationManager(ctx context.Context, client *db.SqlClient) (migrations.Manager, error)
• func TableExists(ctx context.Context, db *sqlx.DB, tableName string, schema string) (bool, error)
• func ViewExists(ctx context.Context, db *sqlx.DB, tableName string, schema string) (bool, error)
• type AdvisoryLock
  • func NewAdvisoryLock(ctx context.Context, db *sqlx.DB, id int) (*AdvisoryLock, error)
  • func (l *AdvisoryLock) Close()
  • func (l *AdvisoryLock) Lock(ctx context.Context) error
  • func (l *AdvisoryLock) TryLock(ctx context.Context) (bool, error)
  • func (l *AdvisoryLock) Unlock(ctx context.Context) error
• type ClientConfig
  • func NewClientConfig() *ClientConfig
  • func (c ClientConfig) Apply(db *sqlx.DB) error
  • func (c ClientConfig) Validate() error
• type LockConn

Constants

const (
	DefaultIdleConns          = 2
	DefaultMaxConns           = 4
	DefaultConnLifeTimeSecond = 3600
	DefaultConnIdleTimeSecond = 1800

	ErrEmptyDSN            = utils.Error("Empty DSN")
	ErrNilConfig           = utils.Error("Config is nil")
	ErrInvalidIdleConns    = utils.Error("Invalid idleConns")
	ErrInvalidMaxConns     = utils.Error("Invalid maxConns")
	ErrInvalidConnLifeTime = utils.Error("connLifeTime must be >= 1")
	ErrInvalidConnIdleTime = utils.Error("connIdleTime must be >= 1")
)

const (
	EngineSchema         = "public"
	MigrationTable       = "db_migration"
	EngineMigrationTable = EngineSchema + "." + MigrationTable

	MigrationLockId = 2343452349
)

const (
	SchemaDefault = "public"

	TblTypeTable        = "BASE TABLE"
	TblTypeView         = "VIEW"
	TblTypeForeignTable = "FOREIGN TABLE"
	TblTypeLocal        = "LOCAL TEMPORARY"
)

Functions

func DialectOptions

func DialectOptions() *goqu.SQLDialectOptions

func ForeignTableExists

func ForeignTableExists(ctx context.Context, db *sqlx.DB, tableName string, schema string) (bool, error)

ForeignTableExists returns true if specified foreign table exists

func GetServerVersion

func GetServerVersion(db *sqlx.DB, ctx context.Context) (string, error)

GetServerVersion fetch postgresql version

func NewClient

func NewClient(config *ClientConfig) (*db.SqlClient, error)

func NewMigrationManager

func NewMigrationManager(ctx context.Context, client *db.SqlClient) (migrations.Manager, error)

func TableExists

func TableExists(ctx context.Context, db *sqlx.DB, tableName string, schema string) (bool, error)

TableExists returns true if specified table exists

func ViewExists

func ViewExists(ctx context.Context, db *sqlx.DB, tableName string, schema string) (bool, error)

ViewExists returns true if specified view exists

Types

type AdvisoryLock

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

func NewAdvisoryLock

func NewAdvisoryLock(ctx context.Context, db *sqlx.DB, id int) (*AdvisoryLock, error)

func (*AdvisoryLock) Close

func (l *AdvisoryLock) Close()

func (*AdvisoryLock) Lock

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

Lock attempts to perform a lock, and waits until it is available

func (*AdvisoryLock) TryLock

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

TryLock attempts to perform a lock, and returns true if operation was successful

func (*AdvisoryLock) Unlock

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

Unlock unlocks a given lock Unlock of a given lock needs to be done with the same connection

type ClientConfig

type ClientConfig struct {
	DSN          string `json:"dsn"`
	MaxOpenConns int    `json:"maxOpenConns"` // MaxOpenConns max number of pool connections
	MaxIdleConns int    `json:"maxIdleConns"` // MaxIdleConns max number of idle pool connections

	// ConnLifeTime is the duration in seconds since creation after which a connection will be automatically closed
	ConnLifetime int `json:"connLifetime"`
	// ConnIdleTime is the duration in seconds after which an idle connection will be automatically closed by the health check
	ConnIdleTime int `json:"connIdleTime"`
}

func NewClientConfig

func NewClientConfig() *ClientConfig

func (ClientConfig) Apply

func (c ClientConfig) Apply(db *sqlx.DB) error

func (ClientConfig) Validate

func (c ClientConfig) Validate() error

type LockConn

type LockConn interface {
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
	Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
}