postgres

package

v0.3.0-prerelease.2 Latest Latest Go to latest Published: Mar 20, 2024 License: MIT Imports: 19 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/get-eventually/go-eventually

Links

Open Source Insights

Documentation ¶

Overview ¶

Package postgres contains implementations of go-eventually interfaces specific to PostgreSQL, such as Aggregate Repository, Event Store, etc.

Index ¶

func RunMigrations(db *sql.DB) error
type AggregateRepository
- func NewAggregateRepository[ID aggregate.ID, T aggregate.Root[ID]](conn *pgxpool.Pool, aggregateType aggregate.Type[ID, T], ...) AggregateRepository[ID, T]
- func (repo AggregateRepository[ID, T]) Get(ctx context.Context, id ID) (T, error)
- func (repo AggregateRepository[ID, T]) Save(ctx context.Context, root T) (err error)
type EventStore
- func NewEventStore(conn *pgxpool.Pool, messageSerde serde.Bytes[message.Message]) EventStore
- func (es EventStore) Append(ctx context.Context, id event.StreamID, expected version.Check, ...) (version.Version, error)
- func (es EventStore) Stream(ctx context.Context, stream event.StreamWrite, id event.StreamID, ...) error

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func RunMigrations ¶

func RunMigrations(db *sql.DB) error

RunMigrations runs the latest migrations for the postgres integration.

Make sure to run these in the entrypoint of your application, ideally before building a postgres interface implementation.

Types ¶

type AggregateRepository ¶

type AggregateRepository[ID aggregate.ID, T aggregate.Root[ID]] struct {
	// contains filtered or unexported fields
}

AggregateRepository implements the aggregate.Repository interface for PostgreSQL databases.

This implementation uses the "aggregates" table in the database as its main operational table. At the same time, it also writes to both "events" and "event_streams" to append the Domain events recorded by Aggregate Roots. These updates are performed within the same transaction.

func NewAggregateRepository ¶

func NewAggregateRepository[ID aggregate.ID, T aggregate.Root[ID]](
	conn *pgxpool.Pool,
	aggregateType aggregate.Type[ID, T],
	aggregateSerde serde.Bytes[T],
	messageSerde serde.Bytes[message.Message],
) AggregateRepository[ID, T]

NewAggregateRepository returns a new AggregateRepository instance.

func (AggregateRepository[ID, T]) Get ¶

func (repo AggregateRepository[ID, T]) Get(ctx context.Context, id ID) (T, error)

Get returns the aggregate.Root instance specified by the provided id. Returns aggregate.ErrRootNotFound if the Aggregate Root doesn't exist.

func (AggregateRepository[ID, T]) Save ¶

func (repo AggregateRepository[ID, T]) Save(ctx context.Context, root T) (err error)

Save saves the new state of the provided aggregate.Root instance.

type EventStore ¶

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

EventStore is an event.Store implementation targeted to PostgreSQL databases.

The implementation uses "event_streams" and "events" as their operational tables. Updates to these tables are transactional.

func NewEventStore ¶

func NewEventStore(conn *pgxpool.Pool, messageSerde serde.Bytes[message.Message]) EventStore

NewEventStore returns a new EventStore instance.

func (EventStore) Append ¶

func (es EventStore) Append(
	ctx context.Context,
	id event.StreamID,
	expected version.Check,
	events ...event.Envelope,
) (version.Version, error)

Append implements event.Store.

func (EventStore) Stream ¶

func (es EventStore) Stream(
	ctx context.Context,
	stream event.StreamWrite,
	id event.StreamID,
	selector version.Selector,
) error

Stream implements the event.Streamer interface.

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
internal Package internal contains utilities and helper functions useful in the scope of eventually's postgres implementation.	Package internal contains utilities and helper functions useful in the scope of eventually's postgres implementation.

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL