README
¶
Postgres
Documentation
¶
Index ¶
- Variables
- type EventStore
- func (store *EventStore) Connect(ctx context.Context) error
- func (store *EventStore) Delete(ctx context.Context, events ...event.Event) error
- func (store *EventStore) Find(ctx context.Context, id uuid.UUID) (event.Event, error)
- func (store *EventStore) Insert(ctx context.Context, events ...event.Event) error
- func (store *EventStore) Pool() *pgxpool.Pool
- func (store *EventStore) Query(ctx context.Context, query event.Query) (<-chan event.Event, <-chan error, error)
- type EventStoreOption
Constants ¶
This section is empty.
Variables ¶
var ErrTableDoesNotExist = errors.New("table does not exist")
ErrTableDoesNotExist is returned when the configured event table does not exist in the connected database.
Functions ¶
This section is empty.
Types ¶
type EventStore ¶
type EventStore struct {
// contains filtered or unexported fields
}
EventStore is a PostgreSQL event store.
func NewEventStore ¶
func NewEventStore(enc codec.Encoding, opts ...EventStoreOption) *EventStore
NewEventStore returns a new PostgreSQL event store. If not otherwise specified using the URL() option, os.Getenv("POSTGRES_EVENTSTORE") is used as the connection string.
On connect, the store looks up the event table (see the Table option). If the table does not exist, the store creates it together with its indexes, which requires the CREATE privilege on the schema. If the table already exists — for example because it was created by a migration tool using a more privileged role — the store instead validates that the table matches the following schema and creates any missing indexes if permitted:
CREATE TABLE events ( id UUID PRIMARY KEY NOT NULL, name VARCHAR(255) NOT NULL, time BIGINT NOT NULL, aggregate_id UUID, aggregate_name VARCHAR(255), aggregate_version INTEGER, data JSONB ); CREATE INDEX goes_name ON events (name); CREATE INDEX goes_time ON events (time); CREATE UNIQUE INDEX goes_aggregate ON events (aggregate_id, aggregate_name, aggregate_version);
TEXT columns are accepted in place of VARCHAR. The UNIQUE index on (aggregate_id, aggregate_name, aggregate_version) is required because it enforces optimistic concurrency for aggregates: Connect fails if it is missing and cannot be created with the privileges of the connected role.
func (*EventStore) Connect ¶
func (store *EventStore) Connect(ctx context.Context) error
Connect connects to the PostgreSQL server. Connect is automatically called from the Insert, Find, Query, and Delete methods if not called explicitly.
func (*EventStore) Pool ¶
func (store *EventStore) Pool() *pgxpool.Pool
Pool returns the underlying Postgres connection pool. Pool must only be called AFTER the connection to Postgres has been established. Otherwise the returned Pool will be nil. It is not thread-safe to call Pool concurrently with Connect.
type EventStoreOption ¶
type EventStoreOption func(*EventStore)
EventStoreOption is an optionn for the PostgreSQL event store.
func Database ¶
func Database(name string) EventStoreOption
Database returns an EventStoreOption that configures the used database. Defaults to "goes".
func Pool ¶ added in v0.6.6
func Pool(pool *pgxpool.Pool) EventStoreOption
Pool returns an EventStoreOption that sets the pool to an existing pgxpool.Pool. The provided pool must already be connected to the PostgreSQL server.
func Table ¶
func Table(name string) EventStoreOption
Table returns an EventStoreOption that configures the used table for events. Defaults to "events".
func URL ¶
func URL(url string) EventStoreOption
URL returns an EventStoreOption that specifies the connection string to the PostgreSQL server.
Source Files