xuu
txt.sour.is
I am Awesome! https://key.sour.is/id/me@sour.is
I am Awesome! https://key.sour.is/id/me@sour.is
⨁ Follow
button on their profile page or use the Follow form and enter a Twtxt URL. You may also find other feeds of interest via Feeds. Welcome! 🤗
AM I HAX0RD?
@lyse@lyse.isobeef.org ¯\_(ツ)_/¯ fixed your arm.
@prologic@twtxt.net correct type parameters. 😅
(cont.)
Just to give some context on some of the components around the code structure.. I wrote this up around an earlier version of aggregate code. This generic bit simplifies things by removing the need of the Crud functions for each aggregate.
Domain ObjectsA domain object can be used as an aggregate by adding the event.AggregateRoot
struct and finish implementing event.Aggregate. The AggregateRoot implements logic for adding events after they are either Raised by a command or Appended by the eventstore Load or service ApplyFn methods. It also tracks the uncommitted events that are saved using the eventstore Save method.
type User struct {
Identity string ```json:"identity"`
CreatedAt time.Time
event.AggregateRoot
}
// StreamID for the aggregate when stored or loaded from ES.
func (a *User) StreamID() string {
return "user-" + a.Identity
}
// ApplyEvent to the aggregate state.
func (a *User) ApplyEvent(lis ...event.Event) {
for _, e := range lis {
switch e := e.(type) {
case *UserCreated:
a.Identity = e.Identity
a.CreatedAt = e.EventMeta().CreatedDate
/* ... */
}
}
}
Events
Events are applied to the aggregate. They are defined by adding the event.Meta
and implementing the getter/setters for event.Event
type UserCreated struct {
eventMeta event.Meta
Identity string
}
func (c *UserCreated) EventMeta() (m event.Meta) {
if c != nil {
m = c.eventMeta
}
return m
}
func (c *UserCreated) SetEventMeta(m event.Meta) {
if c != nil {
c.eventMeta = m
}
}
Reading Events from EventStore
With a domain object that implements the event.Aggregate
the event store client can load events and apply them using the Load(ctx, agg)
method.
// GetUser populates an user from event store.
func (rw *User) GetUser(ctx context.Context, userID string) (*domain.User, error) {
user := &domain.User{Identity: userID}
err := rw.es.Load(ctx, user)
if err != nil {
if err != nil {
if errors.Is(err, eventstore.ErrStreamNotFound) {
return user, ErrNotFound
}
return user, err
}
return nil, err
}
return user, err
}
OnX Commands
An OnX command will validate the state of the domain object can have the command performed on it. If it can be applied it raises the event using event.Raise() Otherwise it returns an error.
// OnCreate raises an UserCreated event to create the user.
// Note: The handler will check that the user does not already exsist.
func (a *User) OnCreate(identity string) error {
event.Raise(a, &UserCreated{Identity: identity})
return nil
}
// OnScored will attempt to score a task.
// If the task is not in a Created state it will fail.
func (a *Task) OnScored(taskID string, score int64, attributes Attributes) error {
if a.State != TaskStateCreated {
return fmt.Errorf("task expected created, got %s", a.State)
}
event.Raise(a, &TaskScored{TaskID: taskID, Attributes: attributes, Score: score})
return nil
}
Crud Operations for OnX Commands
The following functions in the aggregate service can be used to perform creation and updating of aggregates. The Update function will ensure the aggregate exists, where the Create is intended for non-existent aggregates. These can probably be combined into one function.
// Create is used when the stream does not yet exist.
func (rw *User) Create(
ctx context.Context,
identity string,
fn func(*domain.User) error,
) (*domain.User, error) {
session, err := rw.GetUser(ctx, identity)
if err != nil && !errors.Is(err, ErrNotFound) {
return nil, err
}
if err = fn(session); err != nil {
return nil, err
}
_, err = rw.es.Save(ctx, session)
return session, err
}
// Update is used when the stream already exists.
func (rw *User) Update(
ctx context.Context,
identity string,
fn func(*domain.User) error,
) (*domain.User, error) {
session, err := rw.GetUser(ctx, identity)
if err != nil {
return nil, err
}
if err = fn(session); err != nil {
return nil, err
}
_, err = rw.es.Save(ctx, session)
return session, err
}
Progress! so i have moved into working on aggregates. Which are a grouping of events that replayed on an object set the current state of the object. I came up with this little bit of generic wonder.
type PA[T any] interface {
event.Aggregate
*T
}
// Create uses fn to create a new aggregate and store in db.
func Create[A any, T PA[A]](ctx context.Context, es *EventStore, streamID string, fn func(context.Context, T) error) (agg T, err error) {
ctx, span := logz.Span(ctx)
defer span.End()
agg = new(A)
agg.SetStreamID(streamID)
if err = es.Load(ctx, agg); err != nil {
return
}
if err = event.NotExists(agg); err != nil {
return
}
if err = fn(ctx, agg); err != nil {
return
}
var i uint64
if i, err = es.Save(ctx, agg); err != nil {
return
}
span.AddEvent(fmt.Sprint("wrote events = ", i))
return
}
This lets me do something like this:
a, err := es.Create(ctx, r.es, streamID, func(ctx context.Context, agg *domain.SaltyUser) error {
return agg.OnUserRegister(nick, key)
})
I can tell the function the type being modified and returned using the function argument that is passed in. pretty cray cray.
With respect to logging.. oh man.. it really depends on the environment you are working in.. development? log everything! and use a jeager open trace for the super gnarly places. So you can see whats going on while building. But, for production? metrics are king. I don’t want to sift through thousands of lines but have a measure that can tell me the health of the service.
errors.As(…)
just doesn't feel natural. errors.Is(…)
only just. I mainly avoided it. Yesterday evening I actually researched a bit about that and found this article on errors with Go 1.13. It shed a little bit of light, but I still have a long way to go, I reckon.
+1
I have updated my eventDB to have subscriptions! It now has websockets like msgbus. I have also added a in memory store that can be used along side the disk backed wal.
@prologic can Yarn pods be consumers to other yarn pods?
Hi, I am playing with making an event sourcing database. Its super alpha but I thought I would share since others are talking about databases and such.
It’s super basic. Using tidwall/wal as the disk backing. The first use case I am playing with is an implementation of msgbus. I can post events to it and read them back in reverse order.
I plan to expand it to handle other event sourcing type things like aggregates and projections.
Find it here: sour-is/ev