db

Documentation

Index

• func GetTx(ctx context.Context) (pgx.Tx, bool)
• func Transaction(ctx context.Context, dbPool *pgxpool.Pool, ...) error
• func WithTx(ctx context.Context, tx pgx.Tx) context.Context
• type AggregateLoaderError
  • func (ale AggregateLoaderError) Error() string
  • func (ale AggregateLoaderError) Unwrap() error
• type AggregateNotFoundError
  • func (anfe AggregateNotFoundError) Error() string
• type AggregateRepository
  • func NewAggregateRepository(aggregate eventsourcing.Aggregate, db *pgxpool.Pool, ...) *AggregateRepository
  • func (ar *AggregateRepository) AddProjector(event eventsourcing.Event, projector eventsourcing.Projector)
  • func (ar *AggregateRepository) Load(ctx context.Context, aggregateID uuid.UUID) (eventsourcing.Aggregate, error)
  • func (ar *AggregateRepository) NewEmptyAggregate() eventsourcing.Aggregate
  • func (ar *AggregateRepository) Save(ctx context.Context, aggregate eventsourcing.Aggregate) error
• type AggregateRepositoryOption
  • func WithAggregateLoader(aggregateLoader eventsourcing.AggregateLoader) AggregateRepositoryOption
  • func WithAggregateSaver(aggregateSaver eventsourcing.AggregateSaver) AggregateRepositoryOption
• type AggregateSaverError
  • func (ale AggregateSaverError) Error() string
• type ApplyEventError
  • func (aee ApplyEventError) Error() string
• type EventMarshalError
  • func (eue EventMarshalError) Error() string
• type EventModel
  • func NewEventModelFromEvent(event eventsourcing.Event) (*EventModel, error)
  • func (ev *EventModel) ToEvent(reg *eventsourcing.EventRegistry) (eventsourcing.Event, error)
• type EventProjectorError
  • func (epe EventProjectorError) Error() string
• type EventPublishError
  • func (epe EventPublishError) Error() string
• type EventStore
  • func NewEventStore(eventTable string, er *eventsourcing.EventRegistry, dbPool *pgxpool.Pool) *EventStore
  • func (es *EventStore) Append(ctx context.Context, events []eventsourcing.Event) error
  • func (es *EventStore) Load(ctx context.Context, aggregateID uuid.UUID, startVersion int) ([]eventsourcing.Event, error)
  • func (es *EventStore) Migration() string
• type EventUnmarshalError
  • func (eue EventUnmarshalError) Error() string
• type EventVersionConflictError
  • func (cee *EventVersionConflictError) Error() string
• type LoadEventError
  • func (lee LoadEventError) Error() string
• type TransactionError
  • func (te TransactionError) Error() string
  • func (te TransactionError) Unwrap() error

Functions

func GetTx

func GetTx(ctx context.Context) (pgx.Tx, bool)

func Transaction

func Transaction(
	ctx context.Context,
	dbPool *pgxpool.Pool,
	txFunc func(ctx context.Context, tx pgx.Tx) error,
) error

Transaction runs txFunc inside a db transaction.

It's useful when running multiple components in the same transaction. For example:

func (r *MyRepository) Save(agg MyAggregate) error {
   err := Transaction(ctx, r.db, func(ctx, tx) error {
   	querier := r.sqlcQuerier.WithTx(tx)
   	eventStore := r.eventStore.WithTx(tx)
   	...
   })
})

If ctx contains a transaction already, it starts a nested transaction.

If txFunc returns a non-nil error, the transaction will be rollbacked. If it's inside a nested transaction, only the inner transaction will be rollbacked, the outer one is not affeted.

func WithTx

func WithTx(ctx context.Context, tx pgx.Tx) context.Context

WithTx adds a pgx.tx from context

Types

type AggregateLoaderError

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

func (AggregateLoaderError) Error

func (ale AggregateLoaderError) Error() string

func (AggregateLoaderError) Unwrap

func (ale AggregateLoaderError) Unwrap() error

type AggregateNotFoundError

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

func (AggregateNotFoundError) Error

func (anfe AggregateNotFoundError) Error() string

type AggregateRepository

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

func NewAggregateRepository

func NewAggregateRepository(
	aggregate eventsourcing.Aggregate, db *pgxpool.Pool,
	options ...AggregateRepositoryOption,
) *AggregateRepository

func (*AggregateRepository) AddProjector

func (ar *AggregateRepository) AddProjector(
	event eventsourcing.Event,
	projector eventsourcing.Projector,
)

AddProjector add a projector to repository.

func (*AggregateRepository) Load

func (ar *AggregateRepository) Load(
	ctx context.Context, aggregateID uuid.UUID,
) (eventsourcing.Aggregate, error)

Load loads a aggregate by aggregateID.

If a AggregateLoader is provided, it loads from the AggregateLoader, otherwise it loads from the event store.

func (*AggregateRepository) NewEmptyAggregate

func (ar *AggregateRepository) NewEmptyAggregate() eventsourcing.Aggregate

func (*AggregateRepository) Save

func (ar *AggregateRepository) Save(ctx context.Context, aggregate eventsourcing.Aggregate) error

Save saves the aggregate.

After appending the uncommitted events to event store, it run projectors. If a AggregateSaver is provided, it also save the aggregate with the AggregateSaver. If any errors happened, all changes will be rollbacked.

type AggregateRepositoryOption

type AggregateRepositoryOption func(*AggregateRepository)

func WithAggregateLoader

func WithAggregateLoader(aggregateLoader eventsourcing.AggregateLoader) AggregateRepositoryOption

WithAggregateLoader configure a AggregateLoader for the repository.

func WithAggregateSaver

func WithAggregateSaver(aggregateSaver eventsourcing.AggregateSaver) AggregateRepositoryOption

WithAggregateSaver configure a AggregateSaver for the repository.

type AggregateSaverError

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

func (AggregateSaverError) Error

func (ale AggregateSaverError) Error() string

type ApplyEventError

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

func (ApplyEventError) Error

func (aee ApplyEventError) Error() string

type EventMarshalError

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

func (EventMarshalError) Error

func (eue EventMarshalError) Error() string

type EventModel

type EventModel struct {
	CreatedAt   time.Time `gorm:"column:created_at"`
	EventType   string    `gorm:"column:event_type"`
	Payload     string    `gorm:"column:payload"`
	AggregateID uuid.UUID `gorm:"column:aggregate_id"`
	ParentID    uuid.UUID `gorm:"column:parent_id"`
	Version     int       `gorm:"column:version"`
}

func NewEventModelFromEvent

func NewEventModelFromEvent(event eventsourcing.Event) (*EventModel, error)

NewEventModelFromEvent converts a event to EventModel.

func (*EventModel) ToEvent

func (ev *EventModel) ToEvent(
	reg *eventsourcing.EventRegistry,
) (eventsourcing.Event, error)

ToEvent converts a EventModel to Event.

type EventProjectorError

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

func (EventProjectorError) Error

func (epe EventProjectorError) Error() string

type EventPublishError

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

func (EventPublishError) Error

func (epe EventPublishError) Error() string

type EventStore

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

func NewEventStore

func NewEventStore(
	eventTable string, er *eventsourcing.EventRegistry, dbPool *pgxpool.Pool,
) *EventStore

func (*EventStore) Append

func (es *EventStore) Append(ctx context.Context, events []eventsourcing.Event) error

Append inserts events to database.

func (*EventStore) Load

func (es *EventStore) Load(
	ctx context.Context, aggregateID uuid.UUID, startVersion int,
) ([]eventsourcing.Event, error)

func (*EventStore) Migration

func (es *EventStore) Migration() string

Migration returns a sql for creating the event table.

type EventUnmarshalError

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

func (EventUnmarshalError) Error

func (eue EventUnmarshalError) Error() string

type EventVersionConflictError

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

func (*EventVersionConflictError) Error

func (cee *EventVersionConflictError) Error() string

type LoadEventError

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

func (LoadEventError) Error

func (lee LoadEventError) Error() string

type TransactionError

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

func (TransactionError) Error

func (te TransactionError) Error() string

func (TransactionError) Unwrap

func (te TransactionError) Unwrap() error