sql

Documentation

Overview

Package sql implements persistent storage using the postgres database.

Index

• Constants
• Variables
• func Bool(b bool) pgtype.Bool
• func BoolPtr(s *bool) pgtype.Bool
• func Error(err error) error
• func Inet(ip net.IP) pgtype.Inet
• func Int4(s int) pgtype.Int4
• func Int4Ptr(s *int) pgtype.Int4
• func Int8(s int) pgtype.Int8
• func Int8Ptr(s *int) pgtype.Int8
• func JSON(b []byte) pgtype.JSON
• func NewTestDB(t *testing.T) string
• func NoRowsInResultError(err error) bool
• func NullString() pgtype.Text
• func String(s string) pgtype.Text
• func StringPtr(s *string) pgtype.Text
• func Timestamptz(t time.Time) pgtype.Timestamptz
• func TimestamptzPtr(t *time.Time) pgtype.Timestamptz
• func UUID(s uuid.UUID) pgtype.UUID
• type Action
• type DB
  • func New(ctx context.Context, opts Options) (*DB, error)
  • func (db *DB) Conn(ctx context.Context) *pggen.DBQuerier
  • func (db *DB) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
  • func (db *DB) Lock(ctx context.Context, table string, ...) error
  • func (db *DB) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
  • func (db *DB) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults
  • func (db *DB) Tx(ctx context.Context, callback func(context.Context, pggen.Querier) error) error
  • func (db *DB) WaitAndLock(ctx context.Context, id int64, fn func(context.Context) error) (err error)
• type ForwardFunc
• type Listener
  • func NewListener(logger logr.Logger, db pool) *Listener
  • func (b *Listener) RegisterFunc(table string, getter ForwardFunc)
  • func (b *Listener) Start(ctx context.Context) error
  • func (b *Listener) Started() <-chan struct{}
• type Options

Constants

const (
	InsertAction = "INSERT"
	UpdateAction = "UPDATE"
	DeleteAction = "DELETE"
)

const TestDatabaseURL = "OTF_TEST_DATABASE_URL"

Variables

var ErrSubscriptionTerminated = errors.New("broker terminated the subscription")

ErrSubscriptionTerminated is for use by subscribers to indicate that their subscription has been terminated by the broker.

Functions

func Bool

func Bool(b bool) pgtype.Bool

Bool converts a go-boolean into a postgres non-null boolean

func BoolPtr

func BoolPtr(s *bool) pgtype.Bool

BoolPtr converts a go-boolean pointer into a postgres nullable boolean

func Error

func Error(err error) error

func Inet

func Inet(ip net.IP) pgtype.Inet

Inet converts net.IP into the postgres type pgtype.Inet

func Int4

func Int4(s int) pgtype.Int4

Int4 converts a go-int into a postgres non-null int4

func Int4Ptr

func Int4Ptr(s *int) pgtype.Int4

Int4Ptr converts a go-int pointer into a postgres nullable int4

func Int8

func Int8(s int) pgtype.Int8

Int8 converts a go-int into a postgres non-null int8

func Int8Ptr

func Int8Ptr(s *int) pgtype.Int8

Int8Ptr converts a go-int pointer into a postgres nullable int8

func JSON

func JSON(b []byte) pgtype.JSON

JSON converts a []byte into a postgres JSON type

func NewTestDB

func NewTestDB(t *testing.T) string

NewTestDB creates a logical database in postgres for a test and returns a connection string for connecting to the database. The database is dropped upon test completion.

func NoRowsInResultError

func NoRowsInResultError(err error) bool

func NullString

func NullString() pgtype.Text

NullString returns a postgres null string

func String

func String(s string) pgtype.Text

String converts a go-string into a postgres non-null string

func StringPtr

func StringPtr(s *string) pgtype.Text

StringPtr converts a go-string pointer into a postgres nullable string

func Timestamptz

func Timestamptz(t time.Time) pgtype.Timestamptz

Timestamptz converts a go-time into a postgres non-null timestamptz

func TimestamptzPtr

func TimestamptzPtr(t *time.Time) pgtype.Timestamptz

TimestamptzPtr converts a go-time pointer into a postgres nullable timestamptz

func UUID

func UUID(s uuid.UUID) pgtype.UUID

UUID converts a google-go-uuid into a postgres non-null UUID

Types

type Action

type Action string

Action is the action that was carried out on a database table

type DB

type DB struct {
	*pgxpool.Pool // db connection pool
	logr.Logger
}

DB provides access to the postgres db as well as queries generated from SQL

func New

func New(ctx context.Context, opts Options) (*DB, error)

New constructs a new DB connection pool, and migrates the schema to the latest version.

func (*DB) Conn

func (db *DB) Conn(ctx context.Context) *pggen.DBQuerier

Conn provides pre-generated queries

func (*DB) Exec

func (db *DB) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)

Exec acquires a connection from the pool and executes the given SQL. If the context contains a transaction then that is used.

func (*DB) Lock

func (db *DB) Lock(ctx context.Context, table string, fn func(context.Context, pggen.Querier) error) error

func (*DB) QueryRow

func (db *DB) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row

func (*DB) SendBatch

func (db *DB) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults

func (*DB) Tx

func (db *DB) Tx(ctx context.Context, callback func(context.Context, pggen.Querier) error) error

Tx provides the caller with a callback in which all operations are conducted within a transaction.

func (*DB) WaitAndLock

func (db *DB) WaitAndLock(ctx context.Context, id int64, fn func(context.Context) error) (err error)

WaitAndLock obtains an exclusive session-level advisory lock. If another session holds the lock with the given id then it'll wait until the other session releases the lock. The given fn is called once the lock is obtained and when the fn finishes the lock is released.

type ForwardFunc

type ForwardFunc func(ctx context.Context, id string, action Action)

ForwardFunc handles forwarding the id and action onto subscribers.

type Listener

type Listener struct {
	logr.Logger
	// contains filtered or unexported fields
}

Listener listens for postgres events

func NewListener

func NewListener(logger logr.Logger, db pool) *Listener

func (*Listener) RegisterFunc

func (b *Listener) RegisterFunc(table string, getter ForwardFunc)

RegisterFunc registers a function that is capable of converting database events for the given table into an OTF event.

func (*Listener) Start

func (b *Listener) Start(ctx context.Context) error

Start the pubsub daemon; listen to notifications from postgres and forward to local pubsub broker. The listening channel is closed once the broker has started listening; from this point onwards published messages will be forwarded.

func (*Listener) Started

func (b *Listener) Started() <-chan struct{}

type Options

type Options struct {
	Logger     logr.Logger
	ConnString string
}

Options for constructing a DB