storage

package module
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Overview

Package storage provides abstractions for database interactions and default implementations.

Package storage provides abstractions for queue interactions and default implementations.

Index

Constants

View Source 
const (
	DefaultDatabaseKind             = "postgres"
	DefaultDatabaseHost             = "127.0.0.1"
	DefaultDatabasePort             = "5432"
	DefaultDatabaseName             = "andurel"
	DefaultDatabaseUser             = "postgres"
	DefaultDatabasePassword         = "postgres"
	DefaultDatabaseSSLMode          = "disable"
	DefaultApplicationName          = "andurel"
	DefaultConnectTimeout           = 5 * time.Second
	DefaultStatementCacheCapacity   = 512
	DefaultDescriptionCacheCapacity = 512
	DefaultMaxOpenConnections       = 25
	DefaultMaxIdleConnections       = 25
	DefaultConnectionMaxLifetime    = time.Hour
	DefaultConnectionMaxIdleTime    = 30 * time.Minute
)

Variables

View Source 
var (
	ErrBeginTx    = errors.New("could not begin transaction")
	ErrRollbackTx = errors.New("could not rollback transaction")
	ErrCommitTx   = errors.New("could not commit transaction")
)

Functions

func RunMigrations 

func RunMigrations(ctx context.Context, db *sql.DB, migrations fs.FS, migrationDir string) error

RunMigrations applies Goose migrations from an embedded filesystem.

Types

type Config added in v0.2.0 

type Config struct {
	DatabaseKind             string
	Host                     string
	Port                     string
	Name                     string
	User                     string
	Password                 string
	SSLMode                  string
	ApplicationName          string
	ConnectTimeout           time.Duration
	StatementCacheCapacity   int
	DescriptionCacheCapacity int
	MaxOpenConnections       int
	MaxIdleConnections       int
	ConnectionMaxLifetime    time.Duration
	ConnectionMaxIdleTime    time.Duration
	OpenTelemetry            bool
}

Config contains PostgreSQL connection settings. NewPostgres starts with DefaultConfig and applies any options supplied by the application.

func DefaultConfig added in v0.2.0 

func DefaultConfig() Config

DefaultConfig returns development-friendly PostgreSQL defaults.

func (Config) DatabaseURL added in v0.2.0 

func (config Config) DatabaseURL() (string, error)

DatabaseURL returns the PostgreSQL connection URL represented by config.

func (Config) Validate added in v0.2.0 

func (config Config) Validate() error

Validate verifies that all required connection settings are present.

type Connection added in v0.2.0 

type Connection interface {
	Executor() bun.IDB
	DB() *sql.DB
}

Connection exposes the Bun executor and its underlying database/sql pool.

type Executor 

type Executor = bun.IDB

Executor is the query interface satisfied by *bun.DB, *bun.Tx, and *bun.Conn. It remains available for application-owned model and factory APIs.

type InsertQueue 

type InsertQueue interface {
	Insert(context.Context, river.JobArgs, *river.InsertOpts) (*rivertype.JobInsertResult, error)
	InsertTx(context.Context, *sql.Tx, river.JobArgs, *river.InsertOpts) (*rivertype.JobInsertResult, error)
	InsertMany(context.Context, []river.InsertManyParams) ([]*rivertype.JobInsertResult, error)
	InsertManyTx(context.Context, *sql.Tx, []river.InsertManyParams) ([]*rivertype.JobInsertResult, error)
	InsertManyFast(context.Context, []river.InsertManyParams) (int, error)
	InsertManyFastTx(context.Context, *sql.Tx, []river.InsertManyParams) (int, error)
}

InsertQueue is the subset of River used to insert jobs.

type Option added in v0.2.0 

type Option func(*postgresOptions) error

Option configures a PostgreSQL connection.

func WithApplicationName added in v0.2.0 

func WithApplicationName(name string) Option

WithApplicationName sets the PostgreSQL application_name runtime parameter.

func WithConfig added in v0.2.0 

func WithConfig(config Config) Option

WithConfig replaces the default connection settings.

func WithConnectTimeout added in v0.2.0 

func WithConnectTimeout(timeout time.Duration) Option

WithConnectTimeout sets the PostgreSQL connection timeout.

func WithConnectionMaxIdleTime added in v0.2.0 

func WithConnectionMaxIdleTime(idleTime time.Duration) Option

WithConnectionMaxIdleTime sets the maximum time a connection may remain idle.

func WithConnectionMaxLifetime added in v0.2.0 

func WithConnectionMaxLifetime(lifetime time.Duration) Option

WithConnectionMaxLifetime sets the maximum reusable connection lifetime.

func WithDatabaseURL added in v0.2.0 

func WithDatabaseURL(databaseURL string) Option

WithDatabaseURL configures the connection from a PostgreSQL URL.

func WithDescriptionCacheCapacity added in v0.2.0 

func WithDescriptionCacheCapacity(capacity int) Option

WithDescriptionCacheCapacity sets the query description cache capacity.

func WithMaxIdleConnections added in v0.2.0 

func WithMaxIdleConnections(count int) Option

WithMaxIdleConnections sets the database/sql idle connection limit.

func WithMaxOpenConnections added in v0.2.0 

func WithMaxOpenConnections(count int) Option

WithMaxOpenConnections sets the database/sql open connection limit.

func WithOpenTelemetry added in v0.2.0 

func WithOpenTelemetry(config TelemetryConfig) Option

WithOpenTelemetry configures PostgreSQL tracing and metrics.

func WithRuntimeParameters added in v0.2.0 

func WithRuntimeParameters(parameters map[string]string) Option

WithRuntimeParameters merges PostgreSQL runtime parameters.

func WithStatementCacheCapacity added in v0.2.0 

func WithStatementCacheCapacity(capacity int) Option

WithStatementCacheCapacity sets the prepared statement cache capacity.

func WithTLSConfig added in v0.2.0 

func WithTLSConfig(config *tls.Config) Option

WithTLSConfig replaces the TLS configuration used by PostgreSQL.

func WithoutOpenTelemetry added in v0.2.0 

func WithoutOpenTelemetry() Option

WithoutOpenTelemetry disables PostgreSQL tracing and metrics.

type Postgres 

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

Postgres wraps a bun.DB.

func NewPostgres 

func NewPostgres(ctx context.Context, options ...Option) (*Postgres, error)

NewPostgres creates a new database connection using sane defaults and any supplied functional options.

func (*Postgres) BeginTx 

func (p *Postgres) BeginTx(ctx context.Context, opts *sql.TxOptions) (bun.Tx, error)

BeginTx starts a new transaction. Caller is responsible for Commit or Rollback.

func (*Postgres) Close 

func (p *Postgres) Close() error

Close closes the database connection.

func (*Postgres) DB added in v0.2.0 

func (p *Postgres) DB() *sql.DB

DB returns the underlying sql.DB.

func (*Postgres) Executor 

func (p *Postgres) Executor() bun.IDB

Executor returns the Bun query executor.

type QueueInsert added in v0.3.0 

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

QueueInsert is an independently constructible River job inserter.

func NewQueueInsert added in v0.3.0 

func NewQueueInsert(connection Connection, options ...QueueOption) (*QueueInsert, error)

NewQueueInsert creates an insert-only queue client using connection's sql.DB.

func (*QueueInsert) Insert added in v0.3.0 

func (q *QueueInsert) Insert(ctx context.Context, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)

func (*QueueInsert) InsertMany added in v0.3.0 

func (q *QueueInsert) InsertMany(ctx context.Context, params []river.InsertManyParams) ([]*rivertype.JobInsertResult, error)

func (*QueueInsert) InsertManyFast added in v0.3.0 

func (q *QueueInsert) InsertManyFast(ctx context.Context, params []river.InsertManyParams) (int, error)

func (*QueueInsert) InsertManyFastTx added in v0.3.0 

func (q *QueueInsert) InsertManyFastTx(ctx context.Context, tx *sql.Tx, params []river.InsertManyParams) (int, error)

func (*QueueInsert) InsertManyTx added in v0.3.0 

func (q *QueueInsert) InsertManyTx(ctx context.Context, tx *sql.Tx, params []river.InsertManyParams) ([]*rivertype.JobInsertResult, error)

func (*QueueInsert) InsertTx added in v0.3.0 

func (q *QueueInsert) InsertTx(ctx context.Context, tx *sql.Tx, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)

type QueueOption added in v0.3.0 

type QueueOption func(*river.Config)

QueueOption configures the underlying River client.

func WithRiverAdvisoryLockPrefix added in v0.3.0 

func WithRiverAdvisoryLockPrefix(prefix int32) QueueOption

func WithRiverCancelledJobRetentionPeriod added in v0.3.0 

func WithRiverCancelledJobRetentionPeriod(period time.Duration) QueueOption

func WithRiverCompletedJobRetentionPeriod added in v0.3.0 

func WithRiverCompletedJobRetentionPeriod(period time.Duration) QueueOption

func WithRiverConfig added in v0.3.0 

func WithRiverConfig(config river.Config) QueueOption

WithRiverConfig replaces the complete River configuration. Options are applied in order, so field-specific options after this one override it.

func WithRiverDiscardedJobRetentionPeriod added in v0.3.0 

func WithRiverDiscardedJobRetentionPeriod(period time.Duration) QueueOption

func WithRiverErrorHandler added in v0.3.0 

func WithRiverErrorHandler(handler river.ErrorHandler) QueueOption

func WithRiverFetchCooldown added in v0.3.0 

func WithRiverFetchCooldown(cooldown time.Duration) QueueOption

func WithRiverFetchPollInterval added in v0.3.0 

func WithRiverFetchPollInterval(interval time.Duration) QueueOption

func WithRiverHooks added in v0.3.0 

func WithRiverHooks(hooks ...rivertype.Hook) QueueOption

func WithRiverID added in v0.3.0 

func WithRiverID(id string) QueueOption

func WithRiverJobCleanerTimeout added in v0.3.0 

func WithRiverJobCleanerTimeout(timeout time.Duration) QueueOption

func WithRiverJobStuckHandler added in v0.3.0 

func WithRiverJobStuckHandler(handler river.JobStuckHandler) QueueOption

func WithRiverJobStuckThreshold added in v0.3.0 

func WithRiverJobStuckThreshold(threshold time.Duration) QueueOption

func WithRiverJobTimeout added in v0.3.0 

func WithRiverJobTimeout(timeout time.Duration) QueueOption

func WithRiverLogger added in v0.3.0 

func WithRiverLogger(logger *slog.Logger) QueueOption

func WithRiverMaxAttempts added in v0.3.0 

func WithRiverMaxAttempts(maxAttempts int) QueueOption

func WithRiverMiddleware added in v0.3.0 

func WithRiverMiddleware(middleware ...rivertype.Middleware) QueueOption

func WithRiverPeriodicJobs added in v0.3.0 

func WithRiverPeriodicJobs(jobs ...*river.PeriodicJob) QueueOption

func WithRiverPollOnly added in v0.3.0 

func WithRiverPollOnly(pollOnly bool) QueueOption

func WithRiverQueues added in v0.3.0 

func WithRiverQueues(queues map[string]river.QueueConfig) QueueOption

func WithRiverReindexerIndexNames added in v0.3.0 

func WithRiverReindexerIndexNames(names ...string) QueueOption

func WithRiverReindexerSchedule added in v0.3.0 

func WithRiverReindexerSchedule(schedule river.PeriodicSchedule) QueueOption

func WithRiverReindexerTimeout added in v0.3.0 

func WithRiverReindexerTimeout(timeout time.Duration) QueueOption

func WithRiverRescueStuckJobsAfter added in v0.3.0 

func WithRiverRescueStuckJobsAfter(period time.Duration) QueueOption

func WithRiverRetryPolicy added in v0.3.0 

func WithRiverRetryPolicy(policy river.ClientRetryPolicy) QueueOption

func WithRiverSchema added in v0.3.0 

func WithRiverSchema(schema string) QueueOption

func WithRiverSkipUnknownJobCheck added in v0.3.0 

func WithRiverSkipUnknownJobCheck(skip bool) QueueOption

func WithRiverSoftStopTimeout added in v0.3.0 

func WithRiverSoftStopTimeout(timeout time.Duration) QueueOption

func WithRiverTestConfig added in v0.3.0 

func WithRiverTestConfig(test river.TestConfig) QueueOption

func WithRiverTestOnly added in v0.3.0 

func WithRiverTestOnly(testOnly bool) QueueOption

func WithRiverWorkers added in v0.3.0 

func WithRiverWorkers(workers *river.Workers) QueueOption

type QueueProcessor added in v0.3.0 

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

QueueProcessor owns the lifecycle of a River worker client.

func NewQueueProcessor added in v0.3.0 

func NewQueueProcessor(connection Connection, options ...QueueOption) (*QueueProcessor, error)

NewQueueProcessor creates a worker client using connection's sql.DB.

func (*QueueProcessor) Start added in v0.3.0 

func (q *QueueProcessor) Start(ctx context.Context) error

Start starts processing jobs and returns once the processor is ready.

func (*QueueProcessor) Stop added in v0.3.0 

func (q *QueueProcessor) Stop(ctx context.Context) error

Stop gracefully stops processing jobs.

type TelemetryConfig added in v0.2.0 

type TelemetryConfig struct {
	TracerProvider         trace.TracerProvider
	MeterProvider          metric.MeterProvider
	Attributes             []attribute.KeyValue
	TrimSQLInSpanName      bool
	IncludeQueryParameters bool
}

TelemetryConfig configures PostgreSQL OpenTelemetry instrumentation without exposing the underlying database driver.

type TestCluster 

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

TestCluster owns a Postgres test container and can create isolated databases.

func NewTestCluster 

func NewTestCluster(ctx context.Context) (*TestCluster, error)

NewTestCluster starts a Postgres container for a test package.

func (*TestCluster) Close 

func (tc *TestCluster) Close(ctx context.Context) error

Close terminates the Postgres test container.

func (*TestCluster) NewTestDB 

func (tc *TestCluster) NewTestDB(t testing.TB, migrations fs.FS, migrationDir string) Connection

NewTestDB creates a migrated, isolated database for one test.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.