postgres

package
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Jul 15, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package postgres provides pgx-backed transactional persistence for outbox envelopes. Writer methods only accept pgx.Tx so callers cannot accidentally use a pool or connection and lose application-write atomicity.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrPoolRequired         = errors.New("outbox/postgres: pool is required")
	ErrClaimOwnerRequired   = errors.New("outbox/postgres: claim owner is required")
	ErrInvalidClaimLimit    = errors.New("outbox/postgres: claim limit is outside configured bounds")
	ErrInvalidLeaseDuration = errors.New("outbox/postgres: lease duration is outside configured bounds")
	ErrLeaseLost            = errors.New("outbox/postgres: lease is no longer owned")
	ErrInvalidAdminLimit    = errors.New("outbox/postgres: administrative batch is outside configured bounds")
	ErrReplayIDsRequired    = errors.New("outbox/postgres: replay IDs are required")
	ErrReplayRequestedBy    = errors.New("outbox/postgres: replay requester is required")
	ErrReplayReasonRequired = errors.New("outbox/postgres: replay reason is required")
	ErrReplayDuplicateID    = errors.New("outbox/postgres: replay IDs must be unique")
	ErrReplayConflict       = errors.New("outbox/postgres: replay selection contains missing or non-terminal records")
	ErrPruneCutoffRequired  = errors.New("outbox/postgres: prune cutoff is required")
	ErrArchiverRequired     = errors.New("outbox/postgres: archiver is required")
	ErrInvalidMessageState  = errors.New("outbox/postgres: message state is invalid")
	ErrInvalidSerialization = errors.New("outbox/postgres: serialization mode is invalid")
)
View Source 
var (
	ErrEmptyBatch          = errors.New("outbox/postgres: batch is empty")
	ErrBatchTooLarge       = errors.New("outbox/postgres: batch exceeds configured limit")
	ErrInvalidBatchLimit   = errors.New("outbox/postgres: batch limit is invalid")
	ErrTransactionRequired = errors.New("outbox/postgres: caller transaction is required")
)

Functions

func Migrations 

func Migrations() fs.FS

Migrations returns the versioned migration files rooted at their filenames. The returned fs.FS can be consumed by go-migrations or another migration runner without exposing a Goose dependency to applications.

Types

type ArchiveFunc 

type ArchiveFunc func(context.Context, []DeliveredMessage) error

ArchiveFunc adapts a function to Archiver.

func (ArchiveFunc) Archive 

func (archive ArchiveFunc) Archive(ctx context.Context, messages []DeliveredMessage) error

Archive forwards delivered messages to the wrapped function.

type Archiver 

type Archiver interface {
	Archive(context.Context, []DeliveredMessage) error
}

Archiver persists delivered messages before the store deletes them. Implementations must tolerate duplicates because a successful archive can be followed by an ambiguous PostgreSQL commit.

type BacklogStats 

type BacklogStats = outbox.BacklogStats

BacklogStats is the root payload-free backlog summary type.

type Claim 

type Claim struct {
	Envelope    outbox.Envelope
	Owner       string
	LeaseToken  string
	LeasedUntil time.Time
}

Claim is an envelope together with the ownership proof required for every subsequent state transition.

type ClaimRequest 

type ClaimRequest struct {
	Owner         string
	Limit         int
	LeaseDuration time.Duration
	Serialization SerializationMode
}

ClaimRequest describes one bounded lease acquisition.

type DeadArchiveFunc 

type DeadArchiveFunc func(context.Context, []DeadMessage) error

DeadArchiveFunc adapts a function to DeadArchiver.

func (DeadArchiveFunc) ArchiveDead 

func (archive DeadArchiveFunc) ArchiveDead(ctx context.Context, messages []DeadMessage) error

ArchiveDead forwards dead letters to the wrapped function.

type DeadArchiver 

type DeadArchiver interface {
	ArchiveDead(context.Context, []DeadMessage) error
}

DeadArchiver persists dead letters before the store deletes them.

type DeadMessage 

type DeadMessage struct {
	Envelope       outbox.Envelope
	DeadLetteredAt time.Time
	LastError      string
}

DeadMessage contains immutable message data and terminal failure context for a dead-letter archive.

type DeliveredMessage 

type DeliveredMessage struct {
	Envelope    outbox.Envelope
	DeliveredAt time.Time
}

DeliveredMessage contains the immutable message data supplied to an archive-before-delete hook.

type InspectRequest 

type InspectRequest struct {
	State  MessageState
	Topic  string
	Before time.Time
	Limit  int
}

InspectRequest selects a bounded administrative summary batch.

type LeaseRef 

type LeaseRef struct {
	ID    string
	Token string
}

LeaseRef identifies a record and the opaque token for its current lease generation. An expired or replaced token cannot mutate the record.

type MessageState 

type MessageState string

MessageState is a durable outbox state accepted by administrative filters. 

const (
	MessageStatePending   MessageState = "pending"
	MessageStateLeased    MessageState = "leased"
	MessageStateDelivered MessageState = "delivered"
	MessageStateDead      MessageState = "dead"
)

type MessageSummary 

type MessageSummary struct {
	ID             string
	Topic          string
	OrderingKey    string
	IdempotencyKey string
	Attempts       int
	AvailableAt    time.Time
	CreatedAt      time.Time
	UpdatedAt      time.Time
	State          MessageState
	LeaseOwner     *string
	LeasedUntil    *time.Time
	DeliveredAt    *time.Time
	DeadLetteredAt *time.Time
	LastError      *string
}

MessageSummary intentionally excludes payload and metadata.

type ReplayRequest 

type ReplayRequest struct {
	IDs         []string
	RequestedBy string
	Reason      string
	AvailableAt time.Time
}

ReplayRequest is an explicit, audited request to make terminal records publishable again. Replaying can produce duplicates and resets attempts.

type SerializationMode 

type SerializationMode uint8

SerializationMode scopes claim serialization. Ordering-key mode treats an empty key as unordered; topic mode serializes every record in each topic. 

const (
	SerializeNone SerializationMode = iota
	SerializeByOrderingKey
	SerializeByTopic
)

type Store 

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

Store coordinates relay state through PostgreSQL.

func NewStore 

func NewStore(pool *pgxpool.Pool, config StoreConfig) (*Store, error)

NewStore creates a PostgreSQL relay store.

func (*Store) ArchiveAndPruneDead 

func (s *Store) ArchiveAndPruneDead(
	ctx context.Context,
	cutoff time.Time,
	limit int,
	archiver DeadArchiver,
) (ids []string, err error)

ArchiveAndPruneDead archives a bounded dead-letter batch before deletion.

func (*Store) ArchiveAndPruneDelivered 

func (s *Store) ArchiveAndPruneDelivered(
	ctx context.Context,
	cutoff time.Time,
	limit int,
	archiver Archiver,
) (ids []string, err error)

ArchiveAndPruneDelivered locks a bounded delivered batch, invokes archiver, and deletes those records only after archival succeeds. The database transaction remains open during archival so concurrent maintenance workers cannot select the same rows. An ambiguous commit can archive a batch more than once, so archives must be idempotent by envelope ID.

func (*Store) Backlog 

func (s *Store) Backlog(ctx context.Context) (BacklogStats, error)

Backlog returns current state counts and the oldest pending availability time. It is intended for low-frequency health and administrative checks.

func (*Store) Claim 

func (s *Store) Claim(ctx context.Context, request ClaimRequest) ([]Claim, error)

Claim atomically leases available or expired messages. PostgreSQL row locks with SKIP LOCKED make concurrent relay calls return disjoint records.

func (*Store) DeadLetter 

func (s *Store) DeadLetter(ctx context.Context, lease LeaseRef, cause error) error

DeadLetter moves a leased record to its terminal failure state.

func (*Store) ExtendLease 

func (s *Store) ExtendLease(ctx context.Context, lease LeaseRef, duration time.Duration) (time.Time, error)

ExtendLease moves the lease deadline relative to the PostgreSQL clock.

func (*Store) Inspect 

func (s *Store) Inspect(ctx context.Context, request InspectRequest) ([]MessageSummary, error)

Inspect returns bounded payload-free summaries for operator tooling.

func (*Store) MarkDelivered 

func (s *Store) MarkDelivered(ctx context.Context, lease LeaseRef) error

MarkDelivered records publisher success. A publisher success followed by a failure here must be retried and can therefore result in duplicate delivery.

func (*Store) Ping 

func (s *Store) Ping(ctx context.Context) error

Ping verifies that PostgreSQL accepts a round trip.

func (*Store) PruneDead 

func (s *Store) PruneDead(ctx context.Context, cutoff time.Time, limit int) (ids []string, err error)

PruneDead deletes only dead letters older than cutoff in a bounded batch.

func (*Store) PruneDelivered 

func (s *Store) PruneDelivered(ctx context.Context, cutoff time.Time, limit int) (ids []string, err error)

PruneDelivered deletes only delivered records older than cutoff, with a bounded SKIP LOCKED batch so concurrent maintenance workers stay disjoint.

func (*Store) ReleaseLease 

func (s *Store) ReleaseLease(ctx context.Context, lease LeaseRef) error

ReleaseLease makes a currently owned claim immediately available. It is intended for graceful relay cancellation before publication completes.

func (*Store) Replay 

func (s *Store) Replay(ctx context.Context, request ReplayRequest) (ids []string, err error)

Replay atomically resets the selected terminal records and writes one audit row per record. If any ID is missing or non-terminal, no record is changed.

func (*Store) Retry 

func (s *Store) Retry(ctx context.Context, lease LeaseRef, availableAt time.Time, cause error) error

Retry releases a lease and schedules the next eligible claim time.

type StoreConfig 

type StoreConfig struct {
	Schema              string
	Table               string
	MaxClaimBatch       int
	MaxAdminBatch       int
	MaxLeaseDuration    time.Duration
	LeaseTokenGenerator func() (string, error)
	Observer            outbox.Observer
	Clock               func() time.Time
}

StoreConfig bounds relay claims and selects the outbox table.

type Writer 

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

Writer inserts envelopes through caller-owned pgx transactions.

func NewWriter 

func NewWriter(config WriterConfig) (*Writer, error)

NewWriter creates a transactional writer. Empty fields select the default public.outbox_messages table.

func (*Writer) Insert 

func (w *Writer) Insert(ctx context.Context, tx pgx.Tx, envelope outbox.Envelope) error

Insert adds one envelope to the caller's transaction.

func (*Writer) InsertBatch 

func (w *Writer) InsertBatch(ctx context.Context, tx pgx.Tx, envelopes []outbox.Envelope) error

InsertBatch adds every envelope with one statement in the caller's transaction. PostgreSQL therefore accepts all records or none of them.

type WriterConfig 

type WriterConfig struct {
	Schema       string
	Table        string
	Limits       outbox.Limits
	MaxBatchSize int
}

WriterConfig selects the application-owned outbox table.

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.