postgres

README

Event Storage library

Documentation

Index

• type Config
• type Logger
• type Option
  • func WithDSN(dsn string) Option
  • func WithDefaultSlog() Option
  • func WithLogger(logger Logger) Option
  • func WithNoopLogger() Option
  • func WithPool(pool *pgxpool.Pool) Option
  • func WithSlog(log *slog.Logger) Option
  • func WithStartContext(ctx context.Context) Option
• type Storage
  • func New(opts ...Option) (*Storage, error)
  • func (s *Storage) Close() error
  • func (s *Storage) GetStreamIDs(ctx context.Context, streamType string, storeStreamID string, limit int64) ([]string, string, error)
  • func (s *Storage) Read(ctx context.Context, streamType string, streamID string, eventNumber int64) iter.Seq2[es.Event, error]
  • func (s *Storage) Register(streamType string, types ...es.Content) error
  • func (s *Storage) StartPublish(ctx context.Context, w es.Writer) error
  • func (s *Storage) Write(ctx context.Context, streamType string, events iter.Seq2[es.Event, error]) error

Types

type Config

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

type Logger

type Logger interface {
	InfofCtx(ctx context.Context, template string, args ...any)
	ErrorfCtx(ctx context.Context, template string, args ...any)
}

type Option

type Option func(cfg *Config)

func WithDSN

func WithDSN(dsn string) Option

WithDSN creates a Postgresql connection pool using the provided DSN. The connection pool is closed by the Storage.

func WithDefaultSlog

func WithDefaultSlog() Option

func WithLogger

func WithLogger(logger Logger) Option

func WithNoopLogger

func WithNoopLogger() Option

func WithPool

func WithPool(pool *pgxpool.Pool) Option

WithPool uses an existing pgxpool.Pool instance. It is up to the caller to manage the lifecycle of the pool.

func WithSlog

func WithSlog(log *slog.Logger) Option

func WithStartContext

func WithStartContext(ctx context.Context) Option

WithStartContext uses the provided context during initialization.

type Storage

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

func New

func New(opts ...Option) (*Storage, error)

func (*Storage) Close

func (s *Storage) Close() error

func (*Storage) GetStreamIDs

func (s *Storage) GetStreamIDs(ctx context.Context, streamType string, storeStreamID string, limit int64) ([]string, string, error)

func (*Storage) Read

func (s *Storage) Read(ctx context.Context, streamType string, streamID string, eventNumber int64) iter.Seq2[es.Event, error]

func (*Storage) Register

func (s *Storage) Register(streamType string, types ...es.Content) error

func (*Storage) StartPublish

func (s *Storage) StartPublish(ctx context.Context, w es.Writer) error

func (*Storage) Write

func (s *Storage) Write(ctx context.Context, streamType string, events iter.Seq2[es.Event, error]) error