Documentation
¶
Overview ¶
Package es holds an append-only event store. It shares the engine and the transaction of the application, so a stream and an aggregate commit together.
The store is not a drel model. Its primary key is the pair of stream and version, and drel supports one surrogate key for each model. The statements are therefore hand-written SQL.
Index ¶
- Variables
- func CheckpointSchema(table, dialect string) string
- func Schema(table, dialect string) string
- type Checkpoints
- type Iterator
- type Position
- type Record
- type Store
- func (s *Store) Append(ctx context.Context, stream string, expectedVersion int64, events []any) error
- func (s *Store) AppendWithMetadata(ctx context.Context, stream string, expectedVersion int64, events []any, ...) error
- func (s *Store) Read(ctx context.Context, stream string, fromVersion int64) (Iterator, error)
- func (s *Store) ReadAll(ctx context.Context, from Position, limit int) (Iterator, error)
Constants ¶
This section is empty.
Variables ¶
var ErrConcurrency = errors.New("es: the stream is at another version")
ErrConcurrency reports that the stream moved on since the caller read it. The expected version no longer matches the stream. Read the stream again, decide again, and append again.
Functions ¶
func CheckpointSchema ¶
CheckpointSchema returns the CREATE TABLE DDL for a checkpoint table for the given dialect ("postgres" or "sqlite").
The table holds the transaction ID and the global position, because that pair is the read order of ReadAll. The design document named one position column, which cannot resume the read.
func Schema ¶
Schema returns the CREATE TABLE and CREATE INDEX DDL for an event store for the given dialect ("postgres" or "sqlite").
The primary key is the pair of stream and version. It gives optimistic concurrency at no extra cost: a duplicate version raises a unique violation, which Append maps to ErrConcurrency.
On Postgres xact_id holds the transaction ID of the append. ReadAll returns only the rows below the transaction watermark, so no open transaction can insert an event behind the reader. On SQLite xact_id is always 0, because SQLite serialises writers and the insert order is already the commit order.
Types ¶
type Checkpoints ¶
type Checkpoints struct {
// contains filtered or unexported fields
}
Checkpoints records how far each projection has read the log.
A projection writes its read model and saves its checkpoint in one transaction. It can therefore never record progress that it did not make: a failed handler rolls both back, and the next pass reads the same events.
The position is the pair of transaction ID and global position, because that is the read order of ReadAll. One number cannot resume the read.
func NewCheckpoints ¶
func NewCheckpoints(e *drel.Engine, table string) *Checkpoints
NewCheckpoints creates a checkpoint store on one table. The table must exist. CheckpointSchema emits it.
func (*Checkpoints) Load ¶
Load returns the position one projection reached. An unknown projection returns the zero position, which starts the read at the beginning of the log.
Load uses the transaction in the context when one is present, so it reads its own writes. It uses the engine otherwise.
func (*Checkpoints) Reset ¶
func (c *Checkpoints) Reset(ctx context.Context, projection string) error
Reset returns one projection to the beginning of the log, for a replay. Clear the read model in the same transaction, so the replay starts from an empty read model and an empty checkpoint together.
Reset uses the transaction in the context when one is present, and the engine otherwise.
func (*Checkpoints) Save ¶
Save stores the position of one projection inside the transaction that drel.WithTx put in the context. Call it in the same transaction as the read model write, so the two commit together.
Save panics when the context carries no transaction. A checkpoint outside the read model transaction can record progress that the read model never made, so this is a wiring fault.
type Position ¶
Position locates one event in the global log. It is a pair, because the read orders by the transaction ID first and the global position second. A single number cannot resume the read: a transaction that commits late holds a lower global position than one that already reached the reader.
type Record ¶
type Record struct {
StreamID string
Version int64
Position Position
Type string
Payload []byte
Metadata []byte
RecordedAt time.Time
}
Record is one stored event.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store appends events to streams and reads them back.
func (*Store) Append ¶
func (s *Store) Append(ctx context.Context, stream string, expectedVersion int64, events []any) error
Append writes the events to one stream inside the transaction that drel.WithTx put in the context. The events and the application write therefore commit together, or neither commits.
expectedVersion is the version the caller believes the stream holds. A new stream holds 0. The first event of the append takes expectedVersion+1. A version that already exists returns ErrConcurrency, so two writers that read the same state cannot both win.
An empty slice writes nothing and returns nil.
Append panics when the context carries no transaction. Events outside the application transaction give no atomicity at all, so this is a wiring fault.
func (*Store) AppendWithMetadata ¶
func (s *Store) AppendWithMetadata(ctx context.Context, stream string, expectedVersion int64, events []any, metadata map[string]any) error
AppendWithMetadata appends the events and stamps every one of them with the same metadata. Metadata carries the context of the append, for example the correlation ID, the causation ID and the acting user.
func (*Store) Read ¶
Read returns the events of one stream from fromVersion, inclusive, in version order. Use it to rebuild one aggregate.
func (*Store) ReadAll ¶
ReadAll returns the events after from, across every stream, in the read order of the transaction ID and then the global position. Pass the zero Position to start at the beginning, and pass the position of the last record to continue.
On Postgres the read stops at the transaction watermark. An event of a transaction that is still open, and every event of a transaction that started later, stay invisible until that transaction ends. The reader therefore never passes an event that a later call would place behind it. A long write transaction delays the reader by that much.
A projection stores the returned position with the read model it wrote, in one transaction, and passes it back on the next call.
Source Files