#3 global Schema with encoder and decoder

Author: sokool

aggregates.go

@@ -18,29 +18,29 @@ type Aggregate[R Root] struct {
 	// It is kept in memory
 	OnCreate func(string) (R, error)
 
-	// Events
-	Events Schemas
-
 	// OnSession called on command dispatch when Session not exists
 	//OnSession func(R) (Session, error)
 
-	// OnRead when all events are committed to R and state is rebuild from all previously persisted Event
-	OnRead RootFunc[R]
+	// OnLoad when all events are committed to R and state is rebuild from all previously persisted Event
+	OnLoad RootFunc[R]
 
 	// OnRecall
 	//OnRecall func(Session, R) error
 
-	// OnWrite called just before events are persisted to database
-	OnWrite RootFunc[R]
+	// OnSave called just after events are persisted to database
+	OnSave RootFunc[R]
 
-	// OnCommit
+	// OnCommit when new events are committed to a Root
 	OnCommit func(R, Events) error // todo not able to deny it (error)
 
+	// OnRecall
+	OnRecall func(R) time.Time
+
 	// OnCacheCleanup when aggregate is removed from memory
 	OnCacheCleanup RootFunc[R]
 
-	// RecallAfter default
-	//RecallAfter time.Duration
+	// Events
+	Events Schemas
 
 	CleanCacheAfter time.Duration
 
@@ -56,7 +56,7 @@ type Aggregate[R Root] struct {
 	Writer Writer
 
 	// Logger
-	//Log Printer
+	Log Printer
 
 	// memory keeps created Changelog of Aggregate in order to avoid rebuilding
 	// state of each Aggregate everytime when Thread is called
@@ -89,24 +89,30 @@ func (a *Aggregate[R]) Execute(id string, command RootFunc[R]) error {
 }
 
 func (a *Aggregate[R]) Get(id string) (R, error) {
-	d, r, err := a.read(id)
-	if err != nil {
+	var found bool
+	var d RootID
+	var r R
+	var err error
+
+	if err = a.init(); err != nil {
 		return r, err
 	}
 
-	if a.Store == nil {
-		a.Store = NewEventStore()
+	if r, found = a.memory.Get(id); !found {
+		if r, err = a.OnCreate(id); err != nil {
+			return r, err
+		}
 	}
 
-	rw, evs, m := a.Store.ReadWriter(d), make([]Event[any], a.LoadEventsInChunks), 0
+	if d, err = NewRootID(r); err != nil {
+		return r, err
+	}
+
+	rw, evs, m := a.Store.ReadWriter(d), make(Events, a.LoadEventsInChunks), 0
 	for {
 		switch m, err = rw.ReadAt(evs, r.Version()); {
 
 		case err == ErrEndOfStream || err == nil:
-			if m == 0 {
-				return r, nil
-			}
-
 			if failed := a.commit(r, evs[:m]); failed != nil {
 				return r, failed
 			}
@@ -115,13 +121,13 @@ func (a *Aggregate[R]) Get(id string) (R, error) {
 				continue
 			}
 
-			if a.OnRead != nil {
-				if err = a.OnRead(r); err != nil {
+			if a.OnLoad != nil {
+				if err = a.OnLoad(r); err != nil {
 					return r, err
 				}
 			}
 
-			return r, nil
+			return r, a.memory.Set(r.ID(), r)
 
 		default:
 			return r, Err("%s root read failed due %w", r, err)
@@ -135,7 +141,7 @@ func (a *Aggregate[R]) Set(r R) error {
 	}
 
 	events, err := NewEvents(r)
-	if err != nil || len(events) == 0 {
+	if err = events.Extend(r); err != nil || len(events) == 0 {
 		return err
 	}
 
@@ -158,67 +164,44 @@ func (a *Aggregate[R]) Set(r R) error {
 		return err
 	}
 
-	if a.OnWrite != nil {
-		if err = a.OnWrite(r); err != nil {
-			return err
-		}
-	}
-
 	if a.Writer != nil {
 		if _, err = a.Writer.Write(events); err != nil {
 			return err
 		}
 	}
 
-	return nil
-}
-
-func (a *Aggregate[R]) read(id string) (RootID, R, error) {
-	var r, ok = a.memory.Get(id)
-	var d RootID
-	var err error
-
-	if err = a.init(); err != nil {
-		return d, r, err
-	}
-
-	if !ok {
-		if r, err = a.OnCreate(id); err != nil {
-			return d, r, err
+	if a.OnSave != nil {
+		if err = a.OnSave(r); err != nil {
+			return err
 		}
 	}
 
-	if d, err = NewRootID(r); err != nil {
-		return d, r, err
-	}
-
-	return d, r, nil
+	return a.memory.Set(r.ID(), r)
 }
 
-func (a *Aggregate[R]) commit(r R, e []Event[any]) error {
+func (a *Aggregate[R]) commit(r R, e []Event) error {
 	if len(e) == 0 {
 		return nil
 	}
 
-	if a.OnCommit != nil {
-		if err := a.OnCommit(r, e); err != nil {
-			return err
-		}
-	}
-
 	// todo check if given slice of events has correct iteration of sequences, match it with current
 	//      version of R
 	for i := range e {
-		if err := r.Commit(e[i].Body, e[i].CreatedAt); err != nil {
+		if err := r.Commit(e[i].body, e[i].createdAt); err != nil {
 			return err
 		}
 
 		//s += int64(len(e.Body))
 	}
 
+	if a.OnCommit != nil {
+		if err := a.OnCommit(r, e); err != nil {
+			return err
+		}
+	}
 	//a.version = events[len(events)-1].Sequence
 	//a.size += s
-	return a.memory.Set(r.ID(), r)
+	return nil
 }
 
 func (a *Aggregate[R]) init() (err error) {
@@ -229,6 +212,10 @@ func (a *Aggregate[R]) init() (err error) {
 		}
 	}
 
+	if a.Store == nil {
+		a.Store = NewEventStore()
+	}
+
 	if a.memory == nil {
 		a.memory = NewCache[string, R](a.CleanCacheAfter)
 	}
@@ -258,14 +245,18 @@ func (a *Aggregate[R]) Register(in *Domain) (err error) {
 		return err
 	}
 
-	if a.Store == nil {
+	if _, ok := a.Store.(*store); ok {
 		a.Store = in.store
 	}
 
-	if err = in.schemas.Merge(a.Events); err != nil {
+	if err = registry.merge(a.Events, a.Type); err != nil {
 		return err
 	}
 
+	if a.Log == nil {
+		a.Log = in.logger(a.Type)
+	}
+
 	if a.Writer != nil {
 		in.writers.list = append(in.writers.list, a.Writer)
 	}

aggregates_test.go

@@ -35,8 +35,8 @@ func TestAggregates(t *testing.T) {
 
 func NewDomain(t *testing.T) *stream.Domain {
 	return stream.NewDomain(&stream.Configuration{
-		EventStore: func(s *stream.Schemas, l stream.Printer) stream.EventStore {
-			es, err := mysql.NewEventsStore(os.Getenv("MYSQL_EVENT_STORE"), s, l)
+		EventStore: func(l stream.Printer) stream.EventStore {
+			es, err := mysql.NewEventsStore(os.Getenv("MYSQL_EVENT_STORE"), l)
 			if err != nil {
 				t.Fatal(err)
 			}

domain.go

@@ -2,25 +2,26 @@ package stream
 
 import "os"
 
-type RR interface {
+type Registerer interface {
 	Register(*Domain) error
 }
 
 type Domain struct {
 	store   EventStore
-	schemas *Schemas
 	logger  Logger
 	writers *multiWriter
 }
 
 type Configuration struct {
-	EventStore EventStoreFunc
-	Logger     Logger
+	// Logger
+	Logger func(Type) Printer
+
+	// EventStore factory
+	EventStore func(Printer) EventStore // todo func not needed
 }
 
 func NewDomain(c *Configuration) *Domain {
 	s := Domain{
-		schemas: NewSchemas(),
 		store:   NewEventStore(),
 		logger:  NewLogger(os.Stdout, "stream", true).WithTag,
 		writers: &multiWriter{},
@@ -30,13 +31,13 @@ func NewDomain(c *Configuration) *Domain {
 		s.logger = c.Logger
 	}
 	if c.EventStore != nil {
-		s.store = c.EventStore(s.schemas, s.logger("EventStore"))
+		s.store = c.EventStore(s.logger("EventStore"))
 	}
 
 	return &s
 }
 
-func (s *Domain) Register(r ...RR) error {
+func (s *Domain) Register(r ...Registerer) error {
 	for i := range r {
 		if err := r[i].Register(s); err != nil {
 			return err
@@ -45,6 +46,8 @@ func (s *Domain) Register(r ...RR) error {
 	return nil
 }
 
+func (s *Domain) Run() {}
+
 // MultiWriter creates a writer that duplicates its writes to all the
 // provided writers, similar to the Unix tee(1) command.
 //