events

README

Event Store

References

• Building a minimalistic time-sorted event store using a Key-Value Database in Go

Documentation

Index

• Variables
• func CloseIterator(svc Service) endpoint.Endpoint
• func FetchFromIterator(svc Service) endpoint.Endpoint
• func IteratorEndpoint(svc Service) endpoint.Endpoint
• func MinifyMiddleware() endpoint.Middleware
• func NewIteratorEndpoint(svc Service) endpoint.Endpoint
• func StoreEndpoint(svc Service) endpoint.Endpoint
• type Config
  • func (cfg *Config) SetPath(path string)
• type DataType
• type Event
  • func NewEvent(topic string, payload Payload, ids ...ulid.ULID) *Event
  • func (e *Event) Time() time.Time
• type FetchFromIteratorRequest
• type Iterator
• type NewIteratorRequest
• type Payload
  • func NewPayload(data any) Payload
  • func NewPayloadFromBytes(data []byte, raw ...bool) (p Payload, err error)
  • func NewPayloadFromJSON(data json.RawMessage) Payload
  • func (p *Payload) Bytes() ([]byte, bool)
  • func (p *Payload) JSON() (json.RawMessage, bool)
  • func (p *Payload) MarshalJSON() ([]byte, error)
  • func (p *Payload) SetBytes(data []byte)
  • func (p *Payload) SetData(data any)
  • func (p *Payload) SetJSON(data json.RawMessage)
  • func (p *Payload) UnmarshalJSON(data []byte) error
• type Persistence
• type Repository
• type Service
  • func NewService(events Repository) Service
• type ServiceMiddleware
  • func LoggingMiddleware(log *zap.Logger) ServiceMiddleware
• type StorageDriver
• type StoreRequest

Variables

var (
	ErrEmptyPayload     = errors.New("empty payload")
	ErrIteratorNotFound = errors.New("iterator not found")
	ErrInvalidType      = errors.New("invalid type")
)

var (
	ErrTimeout = errors.New("timeout")
)

Functions

func CloseIterator

func CloseIterator(svc Service) endpoint.Endpoint

func FetchFromIterator

func FetchFromIterator(svc Service) endpoint.Endpoint

func IteratorEndpoint

func IteratorEndpoint(svc Service) endpoint.Endpoint

func MinifyMiddleware

func MinifyMiddleware() endpoint.Middleware

func NewIteratorEndpoint

func NewIteratorEndpoint(svc Service) endpoint.Endpoint

func StoreEndpoint

func StoreEndpoint(svc Service) endpoint.Endpoint

Types

type Config

type Config struct {
	Persistence Persistence `yaml:"persistence"`
	Path        string      `yaml:"-"`
}

func (*Config) SetPath

func (cfg *Config) SetPath(path string)

type DataType

type DataType int

const (
	Any DataType = iota
	JSON
	Bytes
)

type Event

type Event struct {
	ID      ulid.ULID `json:"id"`
	Topic   string    `json:"topic"`
	Payload Payload   `json:"payload"`
}

func NewEvent

func NewEvent(topic string, payload Payload, ids ...ulid.ULID) *Event

func (*Event) Time

func (e *Event) Time() time.Time

type FetchFromIteratorRequest

type FetchFromIteratorRequest struct {
	ID    string
	Batch int
}

type Iterator

type Iterator interface {
	ID() string
	Fetch(batch int) ([]*Event, error)
	Close(err error)
	Done() <-chan struct{}
	Err() error
}

type NewIteratorRequest

type NewIteratorRequest struct {
	Topic string    `json:"topic"`
	Since time.Time `json:"since"`
}

type Payload

type Payload struct {
	Data any
	Type DataType
}

func NewPayload

func NewPayload(data any) Payload

func NewPayloadFromBytes

func NewPayloadFromBytes(data []byte, raw ...bool) (p Payload, err error)

func NewPayloadFromJSON

func NewPayloadFromJSON(data json.RawMessage) Payload

func (*Payload) Bytes

func (p *Payload) Bytes() ([]byte, bool)

func (*Payload) JSON

func (p *Payload) JSON() (json.RawMessage, bool)

func (*Payload) MarshalJSON

func (p *Payload) MarshalJSON() ([]byte, error)

func (*Payload) SetBytes

func (p *Payload) SetBytes(data []byte)

func (*Payload) SetData

func (p *Payload) SetData(data any)

func (*Payload) SetJSON

func (p *Payload) SetJSON(data json.RawMessage)

func (*Payload) UnmarshalJSON

func (p *Payload) UnmarshalJSON(data []byte) error

type Persistence

type Persistence struct {
	Driver StorageDriver `yaml:"driver"`
	DSN    string        `yaml:"dsn"`
}

type Repository

type Repository interface {
	Store(e *Event) error
	Iterator(ctx context.Context, since time.Time) (Iterator, error)
	Close() error
}

type Service

type Service interface {
	Up()
	Down()
	Store(topic string, payload Payload, ids ...ulid.ULID) error
	NewIterator(topic string, since time.Time) (string, error)
	Iterator(id string) (Iterator, error)

	// Iterator
	FetchFromIterator(batch int, id string) ([]*Event, error)
	CloseIterator(id string) error
}

func NewService

func NewService(events Repository) Service

type ServiceMiddleware

type ServiceMiddleware func(Service) Service

func LoggingMiddleware

func LoggingMiddleware(log *zap.Logger) ServiceMiddleware

type StorageDriver

type StorageDriver string

const (
	InMem StorageDriver = "inmem"

	BadgerDB StorageDriver = "badger"
	Redis    StorageDriver = "redis"

	InfluxDB   StorageDriver = "influxdb"
	InfluxDBv2 StorageDriver = "influxdb2"

	MySQL      StorageDriver = "mysql"
	SQLServer  StorageDriver = "sqlserver"
	PostgreSQL StorageDriver = "postgres"
	SQLite     StorageDriver = "sqlite"

	// Document Database
	MongoDB StorageDriver = "mongo"
)

type StoreRequest

type StoreRequest struct {
	ID      ulid.ULID `json:"id"`
	Topic   string    `json:"topic"`
	Payload Payload   `json:"payload"`
}