dbpgx

Documentation

Overview

Package dbpgx implements the Conductor's database interface with a PostgreSQL over a jackc/pgx/v5 SQL driver based implementation.

Index

• func DefaultBackoff() backoff.Backoff
• type DB
  • func Open(ctx context.Context, log *slog.Logger, dsn string, maxConns int32, ...) (*DB, error)
  • func (d DB) Close()
  • func (d DB) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
  • func (d DB) ListenEventInserted(ctx context.Context, onReady func(), onEventInserted func(version int64) error) error
  • func (d DB) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
  • func (d DB) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
  • func (d DB) TxRW(ctx context.Context, fn func(context.Context, db.TxRW) error) error
  • func (d DB) TxReadOnly(ctx context.Context, fn func(context.Context, db.TxReadOnly) error) error
• type Tx
  • func (t *Tx) AppendEvent(ctx context.Context, assumeVersion int64, event db.Event) (version int64, err error)
  • func (t *Tx) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
  • func (t *Tx) InitProjectionVersion(ctx context.Context, id int32) (version int64, err error)
  • func (t *Tx) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
  • func (t *Tx) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
  • func (t *Tx) ReadEventAfterVersion(ctx context.Context, afterVersion int64) (e db.Event, err error)
  • func (t *Tx) ReadEventAtVersion(ctx context.Context, version int64) (e db.Event, err error)
  • func (t *Tx) ReadEvents(ctx context.Context, atVersion int64, reverse bool, buffer []db.Event) (read int, err error)
  • func (t *Tx) ReadProjectionVersion(ctx context.Context, id int32) (version int64, err error)
  • func (t *Tx) ReadSystemVersion(ctx context.Context) (version int64, err error)
  • func (t *Tx) SetProjectionVersion(ctx context.Context, id int32, version int64) error

Functions

func DefaultBackoff

func DefaultBackoff() backoff.Backoff

Types

type DB

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

DB is a pgx connection pool that implements the Conductor's DB interface.

func Open

func Open(
	ctx context.Context, log *slog.Logger, dsn string, maxConns int32,
	backoffConf backoff.Backoff,
) (*DB, error)

Open connects to the database using pgx. It will ping and retry until either a successful connection is established or ctx is canceled.

func (DB) Close

func (d DB) Close()

func (DB) Exec

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

func (DB) ListenEventInserted

func (d DB) ListenEventInserted(
	ctx context.Context, onReady func(), onEventInserted func(version int64) error,
) error

func (DB) Query

func (d DB) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)

func (DB) QueryRow

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

func (DB) TxRW

func (d DB) TxRW(
	ctx context.Context, fn func(context.Context, db.TxRW) error,
) error

TxRW starts a new read-write transaction and executes fn inside of it. If fn returns an error or panic occurs, the transaction is rolled back, otherwise it is committed.

func (DB) TxReadOnly

func (d DB) TxReadOnly(
	ctx context.Context, fn func(context.Context, db.TxReadOnly) error,
) error

TxReadOnly starts a new read-only transaction and executes fn inside.

type Tx

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

func (*Tx) AppendEvent

func (t *Tx) AppendEvent(
	ctx context.Context,
	assumeVersion int64,
	event db.Event,
) (version int64, err error)

func (*Tx) Exec

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

func (*Tx) InitProjectionVersion

func (t *Tx) InitProjectionVersion(
	ctx context.Context, id int32,
) (version int64, err error)

func (*Tx) Query

func (t *Tx) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)

func (*Tx) QueryRow

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

func (*Tx) ReadEventAfterVersion

func (t *Tx) ReadEventAfterVersion(
	ctx context.Context, afterVersion int64,
) (e db.Event, err error)

func (*Tx) ReadEventAtVersion

func (t *Tx) ReadEventAtVersion(
	ctx context.Context, version int64,
) (e db.Event, err error)

func (*Tx) ReadEvents

func (t *Tx) ReadEvents(
	ctx context.Context, atVersion int64, reverse bool, buffer []db.Event,
) (read int, err error)

func (*Tx) ReadProjectionVersion

func (t *Tx) ReadProjectionVersion(
	ctx context.Context, id int32,
) (version int64, err error)

func (*Tx) ReadSystemVersion

func (t *Tx) ReadSystemVersion(ctx context.Context) (version int64, err error)

func (*Tx) SetProjectionVersion

func (t *Tx) SetProjectionVersion(ctx context.Context, id int32, version int64) error