postgres

Documentation

Overview

Package postgres provides a PostgreSQL-backed ledger store.

The store uses a single table for all streams. Streams are created implicitly on first append — no DDL is required per stream.

The caller is responsible for opening and closing the *sql.DB. Import a PostgreSQL driver (e.g., github.com/lib/pq) in your application.

Transaction support: pass a *sql.Tx via ledger.WithTx to have the store participate in an external transaction instead of creating its own.

Index

• type Option
  • func WithLogger(l *slog.Logger) Option
  • func WithTable(name string) Option
• type Store
  • func New(ctx context.Context, db *sql.DB, opts ...Option) (*Store, error)
  • func (s *Store) Append(ctx context.Context, stream string, ...) ([]int64, error)
  • func (s *Store) Close(_ context.Context) error
  • func (s *Store) Count(ctx context.Context, stream string) (int64, error)
  • func (s *Store) Health(ctx context.Context) error
  • func (s *Store) ListStreamIDs(ctx context.Context, opts ...ledger.ListOption) ([]string, error)
  • func (s *Store) Read(ctx context.Context, stream string, opts ...ledger.ReadOption) ([]ledger.StoredEntry[int64, json.RawMessage], error)
  • func (s *Store) SetAnnotations(ctx context.Context, stream string, id int64, annotations map[string]*string) error
  • func (s *Store) SetTags(ctx context.Context, stream string, id int64, tags []string) error
  • func (s *Store) Trim(ctx context.Context, stream string, beforeID int64) (int64, error)
  • func (s *Store) Type() string

Types

type Option

type Option func(*options)

Option configures the PostgreSQL store.

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger sets the structured logger. Defaults to slog.Default().

func WithTable

func WithTable(name string) Option

WithTable sets the table name. Defaults to "ledger_entries".

type Store

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

Store is a PostgreSQL ledger store.

func New

func New(ctx context.Context, db *sql.DB, opts ...Option) (*Store, error)

New creates a new PostgreSQL ledger store. The table and indexes are created automatically. Tag GIN indexes are created asynchronously in the background.

func (*Store) Append

func (s *Store) Append(ctx context.Context, stream string, entries ...ledger.RawEntry[json.RawMessage]) ([]int64, error)

Append adds entries to the named stream. Returns IDs of newly appended entries.

func (*Store) Close

func (s *Store) Close(_ context.Context) error

Close marks the store as closed.

func (*Store) Count

func (s *Store) Count(ctx context.Context, stream string) (int64, error)

Count returns the number of entries in the named stream.

func (*Store) Health

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

Health checks database connectivity.

func (*Store) ListStreamIDs

func (s *Store) ListStreamIDs(ctx context.Context, opts ...ledger.ListOption) ([]string, error)

ListStreamIDs returns distinct stream IDs with at least one entry in this store. Results are ascending by stream ID and cursor-paginated via ListAfter/ListLimit.

func (*Store) Read

func (s *Store) Read(ctx context.Context, stream string, opts ...ledger.ReadOption) ([]ledger.StoredEntry[int64, json.RawMessage], error)

Read returns entries from the named stream.

func (*Store) SetAnnotations

func (s *Store) SetAnnotations(ctx context.Context, stream string, id int64, annotations map[string]*string) error

SetAnnotations merges annotations into an entry. Keys with nil values are deleted.

func (*Store) SetTags

func (s *Store) SetTags(ctx context.Context, stream string, id int64, tags []string) error

SetTags replaces all tags on an entry.

func (*Store) Trim

func (s *Store) Trim(ctx context.Context, stream string, beforeID int64) (int64, error)

Trim deletes entries from the named stream with IDs <= beforeID.

func (*Store) Type

func (s *Store) Type() string

Type returns the table name this store is bound to, which represents the entity type for all streams in this store. Intended for logging/tracing.