The Watcher

xuu

txt.sour.is

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
}

fig. 1

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)
})

fig. 2

I can tell the function the type being modified and returned using the function argument that is passed in. pretty cray cray.

xuu

txt.sour.is

14 Aug 22 18:05 UTC

View Thread

(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 Objects
A 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 {
\treturn "agent-" + a.Identity
}
// ApplyEvent to the aggregate state.
func (a *User) ApplyEvent(lis ...event.Event) {
\tfor _, e := range lis {
\t\tswitch e := e.(type) {
\t\tcase *UserCreated:
\t\t\ta.Identity = e.Identity
\t\t\ta.CreatedAt = e.EventMeta().CreatedDate
        /* ... */
\t\t}
\t}
}

## 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 {
\teventMeta event.Meta

\tIdentity string
}

func (c *UserCreated) EventMeta() (m event.Meta) {
\tif c != nil {
\t\tm = c.eventMeta
\t}
\treturn m
}
func (c *UserCreated) SetEventMeta(m event.Meta) {
\tif c != nil {
\t\tc.eventMeta = m
\t}
}

## 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 agent from event store.
func (rw *User) GetUser(ctx context.Context, agentID string) (*domain.User, error) {
\tagent := &domain.User{Identity: agentID}

\terr := rw.es.Load(ctx, agent)
\tif err != nil {
\t\tif err != nil {
\t\t\tif errors.Is(err, eventstore.ErrStreamNotFound) {
\t\t\t\treturn agent, ErrNotFound
\t\t\t}
\t\t\treturn agent, err
\t\t}
\t\treturn nil, err
\t}
\treturn agent, 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 agent.
// Note: The handler will check that the agent 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 {
\tif a.State != TaskStateCreated {
\t\treturn fmt.Errorf("task expected created, got %s", a.State)
\t}
\tevent.Raise(a, &TaskScored{TaskID: taskID, Attributes: attributes, Score: score})
\treturn 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) {
\tsession, err := rw.GetUser(ctx, identity)
\tif err != nil && !errors.Is(err, ErrNotFound) {
\t\treturn nil, err
\t}

\tif err = fn(session); err != nil {
\t\treturn nil, err
\t}

\t_, err = rw.es.Save(ctx, session)

\treturn 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) {
\tsession, err := rw.GetUser(ctx, identity)
\tif err != nil {
\t\treturn nil, err
\t}

\tif err = fn(session); err != nil {
\t\treturn nil, err
\t}

\t_, err = rw.es.Save(ctx, session)
\treturn session, err
}

xuu

txt.sour.is

14 Aug 22 18:05 UTC

View Thread


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 {
\treturn "user-" + a.Identity
}
// ApplyEvent to the aggregate state.
func (a *User) ApplyEvent(lis ...event.Event) {
\tfor _, e := range lis {
\t\tswitch e := e.(type) {
\t\tcase *UserCreated:
\t\t\ta.Identity = e.Identity
\t\t\ta.CreatedAt = e.EventMeta().CreatedDate
        /* ... */
\t\t}
\t}
}

## 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 {
\teventMeta event.Meta

\tIdentity string
}

func (c *UserCreated) EventMeta() (m event.Meta) {
\tif c != nil {
\t\tm = c.eventMeta
\t}
\treturn m
}
func (c *UserCreated) SetEventMeta(m event.Meta) {
\tif c != nil {
\t\tc.eventMeta = m
\t}
}

## 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) {
\tuser := &domain.User{Identity: userID}

\terr := rw.es.Load(ctx, user)
\tif err != nil {
\t\tif err != nil {
\t\t\tif errors.Is(err, eventstore.ErrStreamNotFound) {
\t\t\t\treturn user, ErrNotFound
\t\t\t}
\t\t\treturn user, err
\t\t}
\t\treturn nil, err
\t}
\treturn user, err
}


// 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 {
\tif a.State != TaskStateCreated {
\t\treturn fmt.Errorf("task expected created, got %s", a.State)
\t}
\tevent.Raise(a, &TaskScored{TaskID: taskID, Attributes: attributes, Score: score})
\treturn nil
}


// 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) {
\tsession, err := rw.GetUser(ctx, identity)
\tif err != nil && !errors.Is(err, ErrNotFound) {
\t\treturn nil, err
\t}

\tif err = fn(session); err != nil {
\t\treturn nil, err
\t}

\t_, err = rw.es.Save(ctx, session)

\treturn 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) {
\tsession, err := rw.GetUser(ctx, identity)
\tif err != nil {
\t\treturn nil, err
\t}

\tif err = fn(session); err != nil {
\t\treturn nil, err
\t}

\t_, err = rw.es.Save(ctx, session)
\treturn session, err
}

xuu

txt.sour.is

14 Aug 22 18:05 UTC

View Thread


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
}

xuu

txt.sour.is

14 Aug 22 18:05 UTC

View Thread


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 {
\treturn "agent-" + a.Identity
}
// ApplyEvent to the aggregate state.
func (a *User) ApplyEvent(lis ...event.Event) {
\tfor _, e := range lis {
\t\tswitch e := e.(type) {
\t\tcase *UserCreated:
\t\t\ta.Identity = e.Identity
\t\t\ta.CreatedAt = e.EventMeta().CreatedDate
        /* ... */
\t\t}
\t}
}

## 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 {
\teventMeta event.Meta

\tIdentity string
}

func (c *UserCreated) EventMeta() (m event.Meta) {
\tif c != nil {
\t\tm = c.eventMeta
\t}
\treturn m
}
func (c *UserCreated) SetEventMeta(m event.Meta) {
\tif c != nil {
\t\tc.eventMeta = m
\t}
}

## 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 agent from event store.
func (rw *User) GetUser(ctx context.Context, agentID string) (*domain.User, error) {
\tagent := &domain.User{Identity: agentID}

\terr := rw.es.Load(ctx, agent)
\tif err != nil {
\t\tif err != nil {
\t\t\tif errors.Is(err, eventstore.ErrStreamNotFound) {
\t\t\t\treturn agent, ErrNotFound
\t\t\t}
\t\t\treturn agent, err
\t\t}
\t\treturn nil, err
\t}
\treturn agent, 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 agent.
// Note: The handler will check that the agent 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 {
\tif a.State != TaskStateCreated {
\t\treturn fmt.Errorf("task expected created, got %s", a.State)
\t}
\tevent.Raise(a, &TaskScored{TaskID: taskID, Attributes: attributes, Score: score})
\treturn 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) {
\tsession, err := rw.GetUser(ctx, identity)
\tif err != nil && !errors.Is(err, ErrNotFound) {
\t\treturn nil, err
\t}

\tif err = fn(session); err != nil {
\t\treturn nil, err
\t}

\t_, err = rw.es.Save(ctx, session)

\treturn 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) {
\tsession, err := rw.GetUser(ctx, identity)
\tif err != nil {
\t\treturn nil, err
\t}

\tif err = fn(session); err != nil {
\t\treturn nil, err
\t}

\t_, err = rw.es.Save(ctx, session)
\treturn session, err
}

xuu

txt.sour.is

14 Aug 22 18:05 UTC

View Thread


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
}


// 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
}


// 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
}

xuu@dev.txt.sour.is

dev.txt.sour.is

14 Aug 22 18:05 UTC

View Thread


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 {
\treturn "user-" + a.Identity
}
// ApplyEvent to the aggregate state.
func (a *User) ApplyEvent(lis ...event.Event) {
\tfor _, e := range lis {
\t\tswitch e := e.(type) {
\t\tcase *UserCreated:
\t\t\ta.Identity = e.Identity
\t\t\ta.CreatedAt = e.EventMeta().CreatedDate
        /* ... */
\t\t}
\t}
}

## 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 {
\teventMeta event.Meta

\tIdentity string
}

func (c *UserCreated) EventMeta() (m event.Meta) {
\tif c != nil {
\t\tm = c.eventMeta
\t}
\treturn m
}
func (c *UserCreated) SetEventMeta(m event.Meta) {
\tif c != nil {
\t\tc.eventMeta = m
\t}
}

## 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) {
\tuser := &domain.User{Identity: userID}

\terr := rw.es.Load(ctx, user)
\tif err != nil {
\t\tif err != nil {
\t\t\tif errors.Is(err, eventstore.ErrStreamNotFound) {
\t\t\t\treturn user, ErrNotFound
\t\t\t}
\t\t\treturn user, err
\t\t}
\t\treturn nil, err
\t}
\treturn user, err
}


// 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 {
\tif a.State != TaskStateCreated {
\t\treturn fmt.Errorf("task expected created, got %s", a.State)
\t}
\tevent.Raise(a, &TaskScored{TaskID: taskID, Attributes: attributes, Score: score})
\treturn nil
}


// 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) {
\tsession, err := rw.GetUser(ctx, identity)
\tif err != nil && !errors.Is(err, ErrNotFound) {
\t\treturn nil, err
\t}

\tif err = fn(session); err != nil {
\t\treturn nil, err
\t}

\t_, err = rw.es.Save(ctx, session)

\treturn 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) {
\tsession, err := rw.GetUser(ctx, identity)
\tif err != nil {
\t\treturn nil, err
\t}

\tif err = fn(session); err != nil {
\t\treturn nil, err
\t}

\t_, err = rw.es.Save(ctx, session)
\treturn session, err
}

prologic

twtxt.net

14 Aug 22 22:39 UTC

View Thread

Hmmm 🤔 The problem of course is the code is less readable/understandable 😅 I _never_ thought Go would ever grow generics, but oh well here we are. I guess it'll take me a while to get used to it 😂

prologic

twtxt.net

14 Aug 22 22:39 UTC

View Thread

prologic

twtxt.net

14 Aug 22 22:41 UTC

View Thread

To be fair though, they aren't as bad as Java or C++ 🤣 And technically they're called "Type Parameters" right? 🤔

prologic

twtxt.net

14 Aug 22 22:41 UTC

View Thread

To be fair though, they aren't as bad as Java or C++ 🤣 And technically they're called "Type Parameters" right? 🤔

xuu@dev.txt.sour.is

dev.txt.sour.is

15 Aug 22 00:02 UTC

View Thread

@prologic correct type parameters. 😅

xuu

txt.sour.is

15 Aug 22 00:02 UTC

View Thread

@prologic correct type parameters. 😅

lyse

lyse.isobeef.org

15 Aug 22 19:15 UTC+0200

View Thread

@xuu Congratulations! That's by far the longest twt I've seen. :-) Working fine in tt. I was a bit worried, that lengths of multiple screens would be hard to read or scroll. But magically it just scrolled line by line.

I reckon two returns can be saved in GetUser(…). Oh, on further inspection there are even two nested err != nil checks.

Generics were desperately needed. I'm glad they finally introduced them. I stumbled across them last week and gave them a shot. The syntax is probably not the very best, but I will get used to it eventually.

xuu@dev.txt.sour.is

dev.txt.sour.is

21 Aug 22 06:12 UTC

View Thread

@lyse hah! I cut some out to fit into my pods 4k limit.

Yeah that does studder a bit. To be honest I have no idea what I was thinking there. This excerpt was written a good year ago.

xuu

txt.sour.is

21 Aug 22 06:12 UTC

View Thread

@lyse hah! I cut some out to fit into my pods 4k limit.

Yeah that does studder a bit. To be honest I have no idea what I was thinking there. This excerpt was written a good year ago.

lyse

lyse.isobeef.org

21 Aug 22 08:30 UTC+0200

View Thread

@xuu Just to be extra sure! :-D