zitadel/internal/eventstore/read_model.go
Silvan b5564572bc
feat(eventstore): increase parallel write capabilities (#5940)
This implementation increases parallel write capabilities of the eventstore.
Please have a look at the technical advisories: [05](https://zitadel.com/docs/support/advisory/a10005) and  [06](https://zitadel.com/docs/support/advisory/a10006).
The implementation of eventstore.push is rewritten and stored events are migrated to a new table `eventstore.events2`.
If you are using cockroach: make sure that the database user of ZITADEL has `VIEWACTIVITY` grant. This is used to query events.
2023-10-19 12:19:10 +02:00

51 lines
1.4 KiB
Go

package eventstore
import "time"
// ReadModel is the minimum representation of a read model.
// It implements a basic reducer
// it might be saved in a database or in memory
type ReadModel struct {
AggregateID string `json:"-"`
ProcessedSequence uint64 `json:"-"`
CreationDate time.Time `json:"-"`
ChangeDate time.Time `json:"-"`
Events []Event `json:"-"`
ResourceOwner string `json:"-"`
InstanceID string `json:"-"`
}
// AppendEvents adds all the events to the read model.
// The function doesn't compute the new state of the read model
func (rm *ReadModel) AppendEvents(events ...Event) {
rm.Events = append(rm.Events, events...)
}
// Reduce is the basic implementation of reducer
// If this function is extended the extending function should be the last step
func (rm *ReadModel) Reduce() error {
if len(rm.Events) == 0 {
return nil
}
if rm.AggregateID == "" {
rm.AggregateID = rm.Events[0].Aggregate().ID
}
if rm.ResourceOwner == "" {
rm.ResourceOwner = rm.Events[0].Aggregate().ResourceOwner
}
if rm.InstanceID == "" {
rm.InstanceID = rm.Events[0].Aggregate().InstanceID
}
if rm.CreationDate.IsZero() {
rm.CreationDate = rm.Events[0].CreatedAt()
}
rm.ChangeDate = rm.Events[len(rm.Events)-1].CreatedAt()
rm.ProcessedSequence = rm.Events[len(rm.Events)-1].Sequence()
// all events processed and not needed anymore
rm.Events = nil
rm.Events = []Event{}
return nil
}