ehscylla

README

EventHorizon with ScyllaDB in Golang

Overview

This is an implementation of eventhorizon.EventStore for ScyllaDB.

Installation

go get github.com/arceushui/eh-scylla

Usage

• EventStore

cluster := gocql.NewCluster(os.Getenv("SCYLLA_HOST"))
cluster.Keyspace = os.Getenv("KEYSPACE")
cluster.Timeout = 5 * time.Second
cluster.PoolConfig.HostSelectionPolicy = gocql.TokenAwareHostPolicy(gocql.RoundRobinHostPolicy())

db, err := gocqlx.WrapSession(cluster.CreateSession())
// Create the event store.
store, err := ehscylla.NewEventStore(db)

Documentation

Index

• Variables
• type AggregateEvent
  • func (a AggregateEvent) MarshalBinary() (data []byte, err error)
  • func (a *AggregateEvent) UnmarshalBinary(data []byte) error
• type Encoder
• type EventStore
  • func NewEventStore(db gocqlx.Session) (*EventStore, error)
  • func (s *EventStore) Clear(ctx context.Context) error
  • func (s *EventStore) Close() error
  • func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)
  • func (s *EventStore) LoadFrom(ctx context.Context, id uuid.UUID, version int) ([]eh.Event, error)
  • func (s *EventStore) LoadSnapshot(ctx context.Context, id uuid.UUID) (*eh.Snapshot, error)
  • func (s *EventStore) RenameEvent(ctx context.Context, from, to eh.EventType) error
  • func (s *EventStore) Replace(ctx context.Context, event eh.Event) error
  • func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersion int) error
  • func (s *EventStore) SaveSnapshot(ctx context.Context, id uuid.UUID, snapshot eh.Snapshot) (err error)
• type SnapshotRecord

Variables

var ErrCouldNotCloseDB = errors.New("could not close database session")

ErrCouldNotCloseDB is when the database could not be close.

var ErrCouldNotMarshalEvent = errors.New("could not marshal event")

ErrCouldNotMarshalEvent is when an event could not be marshaled into JSON.

var ErrCouldNotSaveAggregate = errors.New("could not save aggregate")

ErrCouldNotSaveAggregate is when an aggregate could not be saved.

var ErrCouldNotUnmarshalEvent = errors.New("could not unmarshal event")

ErrCouldNotUnmarshalEvent is when an event could not be unmarshalled into a concrete type.

var ErrVersionConflict = errors.New("can not create/update aggregate")

ErrConflictVersion is when a version conflict occurs when saving an aggregate.

var EventTable = table.New(table.Metadata{
	Name: "event_store",
	Columns: []string{
		"event_id",
		"namespace",
		"aggregate_id",
		"aggregate_type",
		"event_type",
		"raw_event_data",
		"timestamp",
		"version",
		"raw_meta_data",
	},
	PartKey: []string{
		"aggregate_id",
	},
	SortKey: []string{
		"version",
	},
})

var NewUUID = uuid.New

NewUUID for mocking in tests

var SnapshotTable = table.New(table.Metadata{
	Name: "snapshots",
	Columns: []string{
		"aggregate_id",
		"aggregate_type",
		"timestamp",
		"version",
		"state",
	},
	PartKey: []string{
		"aggregate_id",
	},
	SortKey: []string{
		"version",
	},
})

Types

type AggregateEvent

type AggregateEvent struct {
	EventID       [16]byte         `db:"event_id"`
	Namespace     string           `db:"namespace"`
	AggregateID   [16]byte         `db:"aggregate_id"`
	AggregateType eh.AggregateType `db:"aggregate_type"`
	EventType     eh.EventType     `db:"event_type"`
	RawEventData  json.RawMessage  `db:"raw_event_data"`
	Timestamp     time.Time        `db:"timestamp"`
	Version       int              `db:"version"`

	RawMetaData json.RawMessage `db:"raw_meta_data"`
	// contains filtered or unexported fields
}

func (AggregateEvent) MarshalBinary

func (a AggregateEvent) MarshalBinary() (data []byte, err error)

func (*AggregateEvent) UnmarshalBinary

func (a *AggregateEvent) UnmarshalBinary(data []byte) error

type Encoder

type Encoder interface {
	Marshal(eh.EventData) ([]byte, error)
	Unmarshal(eh.EventType, []byte) (eh.EventData, error)
	String() string
}

type EventStore

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

EventStore implements an eh.EventStore for PostgreSQL.

func NewEventStore

func NewEventStore(db gocqlx.Session) (*EventStore, error)

NewEventStore creates a new EventStore.

func (*EventStore) Clear

func (s *EventStore) Clear(ctx context.Context) error

Clear clears the event storage.

func (*EventStore) Close

func (s *EventStore) Close() error

func (*EventStore) Load

func (s *EventStore) Load(ctx context.Context, id uuid.UUID) ([]eh.Event, error)

Load implements the Load method of the eventhorizon.EventStore interface.

func (*EventStore) LoadFrom

func (s *EventStore) LoadFrom(ctx context.Context, id uuid.UUID, version int) ([]eh.Event, error)

LoadFrom loads all events from version for the aggregate id from the store.

func (*EventStore) LoadSnapshot

func (s *EventStore) LoadSnapshot(ctx context.Context, id uuid.UUID) (*eh.Snapshot, error)

func (*EventStore) RenameEvent

func (s *EventStore) RenameEvent(ctx context.Context, from, to eh.EventType) error

RenameEvent implements the RenameEvent method of the eventhorizon.EventStore interface.

func (*EventStore) Replace

func (s *EventStore) Replace(ctx context.Context, event eh.Event) error

Replace implements the Replace method of the eventhorizon.EventStore interface.

func (*EventStore) Save

func (s *EventStore) Save(ctx context.Context, events []eh.Event, originalVersion int) error

Save implements the Save method of the eventhorizon.EventStore interface.

func (*EventStore) SaveSnapshot

func (s *EventStore) SaveSnapshot(ctx context.Context, id uuid.UUID, snapshot eh.Snapshot) (err error)

type SnapshotRecord

type SnapshotRecord struct {
	AggregateID   [16]byte         `db:"aggregate_id"`
	Timestamp     time.Time        `db:"timestamp"`
	Version       int              `db:"version"`
	AggregateType eh.AggregateType `db:"aggregate_type"`
	State         []byte           `db:"state"`
}