postgres

package
v2.5.0 Latest Latest
Warning

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

 Go to latest Published: Feb 25, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package postgres provides PostgreSQL adapters for outbox repository contracts.

Migration files under migrations/ include two mutually exclusive tracks: - schema-per-tenant in migrations/ - column-per-tenant in migrations/column/ Choose one strategy per deployment.

SchemaResolver enforces non-empty tenant context by default. Use WithAllowEmptyTenant only for explicit single-tenant/public-schema flows.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrConnectionRequired        = errors.New("postgres connection is required")
	ErrTransactionRequired       = errors.New("postgres transaction is required")
	ErrStateTransitionConflict   = errors.New("outbox event state transition conflict")
	ErrRepositoryNotInitialized  = errors.New("outbox repository not initialized")
	ErrLimitMustBePositive       = errors.New("limit must be greater than zero")
	ErrIDRequired                = errors.New("id is required")
	ErrAggregateIDRequired       = errors.New("aggregate id is required")
	ErrMaxAttemptsMustBePositive = errors.New("maxAttempts must be greater than zero")
	ErrEventTypeRequired         = errors.New("event type is required")
	ErrTenantResolverRequired    = errors.New("tenant resolver is required")
	ErrTenantDiscovererRequired  = errors.New("tenant discoverer is required")
	ErrNoPrimaryDB               = errors.New("no primary database configured for tenant transaction")
	ErrInvalidIdentifier         = errors.New("invalid sql identifier")
)
View Source 
var ErrDefaultTenantIDInvalid = errors.New("default tenant id must be UUID when tenant is required")

Functions

This section is empty.

Types

type ColumnResolver 

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

ColumnResolver supports column-per-tenant strategy.

ApplyTenant is a no-op because tenant scoping is handled by SQL WHERE clauses in Repository when tenantColumn is configured.

func NewColumnResolver 

func NewColumnResolver(client *libPostgres.Client, opts ...ColumnResolverOption) (*ColumnResolver, error)

func (*ColumnResolver) ApplyTenant 

func (resolver *ColumnResolver) ApplyTenant(_ context.Context, _ *sql.Tx, _ string) error

func (*ColumnResolver) DiscoverTenants 

func (resolver *ColumnResolver) DiscoverTenants(ctx context.Context) ([]string, error)

func (*ColumnResolver) RequiresTenant added in v2.5.0 

func (resolver *ColumnResolver) RequiresTenant() bool

RequiresTenant returns true because column-per-tenant strategy always requires a tenant ID to scope queries via WHERE clauses.

func (*ColumnResolver) TenantColumn 

func (resolver *ColumnResolver) TenantColumn() string

type ColumnResolverOption 

type ColumnResolverOption func(*ColumnResolver)

func WithColumnResolverTableName 

func WithColumnResolverTableName(tableName string) ColumnResolverOption

func WithColumnResolverTenantColumn 

func WithColumnResolverTenantColumn(tenantColumn string) ColumnResolverOption

func WithColumnResolverTenantDiscoveryTTL 

func WithColumnResolverTenantDiscoveryTTL(ttl time.Duration) ColumnResolverOption

type Option 

type Option func(*Repository)

func WithLogger 

func WithLogger(logger libLog.Logger) Option

func WithTableName 

func WithTableName(tableName string) Option

func WithTenantColumn 

func WithTenantColumn(tenantColumn string) Option

func WithTransactionTimeout 

func WithTransactionTimeout(timeout time.Duration) Option

type Repository 

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

Repository persists outbox events in PostgreSQL.

func NewRepository 

func NewRepository(
	client *libPostgres.Client,
	tenantResolver outbox.TenantResolver,
	tenantDiscoverer outbox.TenantDiscoverer,
	opts ...Option,
) (*Repository, error)

NewRepository creates a PostgreSQL outbox repository.

func (*Repository) Create 

func (repo *Repository) Create(ctx context.Context, event *outbox.OutboxEvent) (*outbox.OutboxEvent, error)

Create stores a new outbox event using a new transaction.

func (*Repository) CreateWithTx 

func (repo *Repository) CreateWithTx(
	ctx context.Context,
	tx outbox.Tx,
	event *outbox.OutboxEvent,
) (*outbox.OutboxEvent, error)

CreateWithTx stores a new outbox event using an existing transaction.

func (*Repository) GetByID 

func (repo *Repository) GetByID(ctx context.Context, id uuid.UUID) (*outbox.OutboxEvent, error)

GetByID retrieves an outbox event by id.

func (*Repository) ListFailedForRetry 

func (repo *Repository) ListFailedForRetry(
	ctx context.Context,
	limit int,
	failedBefore time.Time,
	maxAttempts int,
) ([]*outbox.OutboxEvent, error)

ListFailedForRetry lists failed events eligible for retry.

func (*Repository) ListPending 

func (repo *Repository) ListPending(ctx context.Context, limit int) ([]*outbox.OutboxEvent, error)

ListPending retrieves pending outbox events up to the given limit.

func (*Repository) ListPendingByType 

func (repo *Repository) ListPendingByType(
	ctx context.Context,
	eventType string,
	limit int,
) ([]*outbox.OutboxEvent, error)

ListPendingByType retrieves pending outbox events filtered by event type.

func (*Repository) ListTenants 

func (repo *Repository) ListTenants(ctx context.Context) ([]string, error)

ListTenants returns tenant IDs discovered by the configured discoverer.

func (*Repository) MarkFailed 

func (repo *Repository) MarkFailed(ctx context.Context, id uuid.UUID, errMsg string, maxAttempts int) error

MarkFailed marks an outbox event as failed and may transition to invalid.

func (*Repository) MarkInvalid 

func (repo *Repository) MarkInvalid(ctx context.Context, id uuid.UUID, errMsg string) error

MarkInvalid marks an outbox event as invalid.

func (*Repository) MarkPublished 

func (repo *Repository) MarkPublished(ctx context.Context, id uuid.UUID, publishedAt time.Time) error

MarkPublished marks an outbox event as published.

func (*Repository) RequiresTenant 

func (repo *Repository) RequiresTenant() bool

RequiresTenant reports whether repository operations require a tenant ID.

func (*Repository) ResetForRetry 

func (repo *Repository) ResetForRetry(
	ctx context.Context,
	limit int,
	failedBefore time.Time,
	maxAttempts int,
) ([]*outbox.OutboxEvent, error)

ResetForRetry atomically selects and resets failed events to processing.

func (*Repository) ResetStuckProcessing 

func (repo *Repository) ResetStuckProcessing(
	ctx context.Context,
	limit int,
	processingBefore time.Time,
	maxAttempts int,
) ([]*outbox.OutboxEvent, error)

ResetStuckProcessing reclaims long-running processing events.

type SchemaResolver 

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

SchemaResolver applies schema-per-tenant scoping and tenant discovery.

func NewSchemaResolver 

func NewSchemaResolver(client *libPostgres.Client, opts ...SchemaResolverOption) (*SchemaResolver, error)

func (*SchemaResolver) ApplyTenant 

func (resolver *SchemaResolver) ApplyTenant(ctx context.Context, tx *sql.Tx, tenantID string) error

ApplyTenant scopes the current transaction to tenant search_path.

Security invariant: tenantID must remain UUID-validated and identifier-quoted before query construction. This method intentionally relies on both checks to keep dynamic search_path assignment safe.

func (*SchemaResolver) DiscoverTenants 

func (resolver *SchemaResolver) DiscoverTenants(ctx context.Context) ([]string, error)

DiscoverTenants returns tenants by inspecting schema names.

func (*SchemaResolver) RequiresTenant 

func (resolver *SchemaResolver) RequiresTenant() bool

type SchemaResolverOption 

type SchemaResolverOption func(*SchemaResolver)

func WithAllowEmptyTenant 

func WithAllowEmptyTenant() SchemaResolverOption

WithAllowEmptyTenant permits ApplyTenant calls with empty tenant IDs.

Use only in trusted single-tenant contexts where falling back to `public` is explicitly intended. The secure default requires tenant IDs.

func WithDefaultTenantID 

func WithDefaultTenantID(tenantID string) SchemaResolverOption

func WithRequireTenant 

func WithRequireTenant() SchemaResolverOption

WithRequireTenant enforces that every ApplyTenant call receives a non-empty tenant ID. This is the default behavior.

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.