db

Documentation

Index

• func Listen[T Notifiable](ctx context.Context, config *ConnectConfig, notifiables []T) error
• func UnlockThread(ctx context.Context, conn Conn, lockId int) error
• type CallbackAction
  • func NewCallbackAction(callbackName string, handlerName string, handlerType string, ...) *CallbackAction
  • func (cba *CallbackAction) Insert(ctx context.Context, conn Conn) (uuid.UUID, error)
• type Conn
• type ConnectConfig
  • func (conf *ConnectConfig) Connect(ctx context.Context) (*pgx.Conn, error)
  • func (c ConnectConfig) String() string
• type Event
  • func (s *Event) Insert(ctx context.Context, conn Conn) error
• type Notifiable
• type PoolConfig
  • func (conf *PoolConfig) Connect(ctx context.Context) (*pgxpool.Pool, error)
• type Thread
  • func GetProcessableThreads(ctx context.Context, conn Conn, handlerName string, limit int, ...) ([]*Thread, error)
  • func (t *Thread) InsertBackoffEvent(ctx context.Context, conn Conn, retrySeconds int, errorMsg string) error
  • func (t *Thread) InsertFailedEvent(ctx context.Context, conn Conn, errorMsg string) error
  • func (t *Thread) InsertQueuedEvent(ctx context.Context, conn Conn) error
  • func (t *Thread) InsertRetriesExhaustedEvent(ctx context.Context, conn Conn, errorMsg string) error
  • func (t *Thread) InsertSuccessfulEvent(ctx context.Context, conn Conn) error
  • func (t *Thread) Unlock(ctx context.Context, conn Conn) error
• type WorkflowNameQuery
  • func (wnq *WorkflowNameQuery) Exec(ctx context.Context, conn Conn) (string, error)

Functions

func Listen

func Listen[T Notifiable](ctx context.Context, config *ConnectConfig, notifiables []T) error

func UnlockThread

func UnlockThread(ctx context.Context, conn Conn, lockId int) error

Types

type CallbackAction

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

func NewCallbackAction

func NewCallbackAction(
	callbackName string,
	handlerName string,
	handlerType string,
	workflowUuid uuid.UUID,
) *CallbackAction

func (*CallbackAction) Insert

func (cba *CallbackAction) Insert(ctx context.Context, conn Conn) (uuid.UUID, error)

type Conn

type Conn interface {
	Begin(ctx context.Context) (pgx.Tx, error)
	Exec(ctx context.Context, sql string, arguments ...any) (commandTag pgconn.CommandTag, err error)
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

type ConnectConfig

type ConnectConfig struct {
	// we could in the future support more config parameters, see
	// https://pkg.go.dev/github.com/jackc/pgx/v5/pgxpool#Config
	Database *string
}

func (*ConnectConfig) Connect

func (conf *ConnectConfig) Connect(ctx context.Context) (*pgx.Conn, error)

func (ConnectConfig) String

func (c ConnectConfig) String() string

type Event

type Event struct {
	ActionUuid   uuid.UUID
	Time         time.Time
	Status       states.ActionState
	ErrorMsg     string
	RetrySeconds int
}

func (*Event) Insert

func (s *Event) Insert(ctx context.Context, conn Conn) error

type Notifiable

type Notifiable interface {
	GetName() string
	Notify()
}

type PoolConfig

type PoolConfig struct {
	ConnectConfig
}

func (*PoolConfig) Connect

func (conf *PoolConfig) Connect(ctx context.Context) (*pgxpool.Pool, error)

type Thread

type Thread struct {
	Uuid        uuid.UUID
	ActionName  *string
	HandlerName string
	Priority    int
	LockId      int
}

func GetProcessableThreads

func GetProcessableThreads(
	ctx context.Context,
	conn Conn,
	handlerName string,
	limit int,
	ignored []uuid.UUID,
) ([]*Thread, error)

func (*Thread) InsertBackoffEvent

func (t *Thread) InsertBackoffEvent(
	ctx context.Context,
	conn Conn,
	retrySeconds int,
	errorMsg string,
) error

func (*Thread) InsertFailedEvent

func (t *Thread) InsertFailedEvent(ctx context.Context, conn Conn, errorMsg string) error

func (*Thread) InsertQueuedEvent

func (t *Thread) InsertQueuedEvent(ctx context.Context, conn Conn) error

func (*Thread) InsertRetriesExhaustedEvent

func (t *Thread) InsertRetriesExhaustedEvent(ctx context.Context, conn Conn, errorMsg string) error

func (*Thread) InsertSuccessfulEvent

func (t *Thread) InsertSuccessfulEvent(ctx context.Context, conn Conn) error

func (*Thread) Unlock

func (t *Thread) Unlock(ctx context.Context, conn Conn) error

type WorkflowNameQuery

type WorkflowNameQuery struct {
	WorkflowUuid uuid.UUID
}

func (*WorkflowNameQuery) Exec

func (wnq *WorkflowNameQuery) Exec(ctx context.Context, conn Conn) (string, error)