mirror of
https://github.com/zitadel/zitadel.git
synced 2025-08-12 01:37:31 +00:00
fix: move v2 pkgs (#1331)
* fix: move eventstore pkgs * fix: move eventstore pkgs * fix: remove v2 view * fix: remove v2 view
This commit is contained in:
Fabi
@@ -2,74 +2,223 @@ package eventstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"sync"
|
||||
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore/internal/repository"
|
||||
"github.com/caos/zitadel/internal/eventstore/models"
|
||||
es_v2 "github.com/caos/zitadel/internal/eventstore/v2"
|
||||
"github.com/caos/zitadel/internal/eventstore/repository"
|
||||
)
|
||||
|
||||
type Eventstore interface {
|
||||
AggregateCreator() *models.AggregateCreator
|
||||
Health(ctx context.Context) error
|
||||
PushAggregates(ctx context.Context, aggregates ...*models.Aggregate) error
|
||||
FilterEvents(ctx context.Context, searchQuery *models.SearchQuery) (events []*models.Event, err error)
|
||||
LatestSequence(ctx context.Context, searchQuery *models.SearchQueryFactory) (uint64, error)
|
||||
V2() *es_v2.Eventstore
|
||||
Subscribe(aggregates ...models.AggregateType) *Subscription
|
||||
//Eventstore abstracts all functions needed to store valid events
|
||||
// and filters the stored events
|
||||
type Eventstore struct {
|
||||
repo repository.Repository
|
||||
interceptorMutex sync.Mutex
|
||||
eventInterceptors map[EventType]eventTypeInterceptors
|
||||
}
|
||||
|
||||
var _ Eventstore = (*eventstore)(nil)
|
||||
|
||||
type eventstore struct {
|
||||
repo repository.Repository
|
||||
aggregateCreator *models.AggregateCreator
|
||||
|
||||
esV2 *es_v2.Eventstore
|
||||
type eventTypeInterceptors struct {
|
||||
eventMapper func(*repository.Event) (EventReader, error)
|
||||
}
|
||||
|
||||
func (es *eventstore) AggregateCreator() *models.AggregateCreator {
|
||||
return es.aggregateCreator
|
||||
func NewEventstore(repo repository.Repository) *Eventstore {
|
||||
return &Eventstore{
|
||||
repo: repo,
|
||||
eventInterceptors: map[EventType]eventTypeInterceptors{},
|
||||
interceptorMutex: sync.Mutex{},
|
||||
}
|
||||
}
|
||||
|
||||
func (es *eventstore) PushAggregates(ctx context.Context, aggregates ...*models.Aggregate) (err error) {
|
||||
for _, aggregate := range aggregates {
|
||||
if len(aggregate.Events) == 0 {
|
||||
return errors.ThrowInvalidArgument(nil, "EVENT-cNhIj", "no events in aggregate")
|
||||
//Health checks if the eventstore can properly work
|
||||
// It checks if the repository can serve load
|
||||
func (es *Eventstore) Health(ctx context.Context) error {
|
||||
return es.repo.Health(ctx)
|
||||
}
|
||||
|
||||
//PushEvents pushes the events in a single transaction
|
||||
// an event needs at least an aggregate
|
||||
func (es *Eventstore) PushEvents(ctx context.Context, pushEvents ...EventPusher) ([]EventReader, error) {
|
||||
events, constraints, err := eventsToRepository(pushEvents)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
err = es.repo.Push(ctx, events, constraints...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return es.mapEvents(events)
|
||||
}
|
||||
|
||||
func eventsToRepository(pushEvents []EventPusher) (events []*repository.Event, constraints []*repository.UniqueConstraint, err error) {
|
||||
events = make([]*repository.Event, len(pushEvents))
|
||||
for i, event := range pushEvents {
|
||||
data, err := eventData(event)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
for _, event := range aggregate.Events {
|
||||
if err = event.Validate(); err != nil {
|
||||
return errors.ThrowInvalidArgument(err, "EVENT-tzIhl", "validate event failed")
|
||||
}
|
||||
events[i] = &repository.Event{
|
||||
AggregateID: event.Aggregate().ID,
|
||||
AggregateType: repository.AggregateType(event.Aggregate().Typ),
|
||||
ResourceOwner: event.Aggregate().ResourceOwner,
|
||||
EditorService: event.EditorService(),
|
||||
EditorUser: event.EditorUser(),
|
||||
Type: repository.EventType(event.Type()),
|
||||
Version: repository.Version(event.Aggregate().Version),
|
||||
Data: data,
|
||||
}
|
||||
if len(event.UniqueConstraints()) > 0 {
|
||||
constraints = append(constraints, uniqueConstraintsToRepository(event.UniqueConstraints())...)
|
||||
}
|
||||
}
|
||||
err = es.repo.PushAggregates(ctx, aggregates...)
|
||||
|
||||
return events, constraints, nil
|
||||
}
|
||||
|
||||
func uniqueConstraintsToRepository(constraints []*EventUniqueConstraint) (uniqueConstraints []*repository.UniqueConstraint) {
|
||||
uniqueConstraints = make([]*repository.UniqueConstraint, len(constraints))
|
||||
for i, constraint := range constraints {
|
||||
uniqueConstraints[i] = &repository.UniqueConstraint{
|
||||
UniqueType: constraint.UniqueType,
|
||||
UniqueField: constraint.UniqueField,
|
||||
Action: uniqueConstraintActionToRepository(constraint.Action),
|
||||
ErrorMessage: constraint.ErrorMessage,
|
||||
}
|
||||
}
|
||||
return uniqueConstraints
|
||||
}
|
||||
|
||||
//FilterEvents filters the stored events based on the searchQuery
|
||||
// and maps the events to the defined event structs
|
||||
func (es *Eventstore) FilterEvents(ctx context.Context, queryFactory *SearchQueryBuilder) ([]EventReader, error) {
|
||||
query, err := queryFactory.build()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
events, err := es.repo.Filter(ctx, query)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return es.mapEvents(events)
|
||||
}
|
||||
|
||||
func (es *Eventstore) mapEvents(events []*repository.Event) (mappedEvents []EventReader, err error) {
|
||||
mappedEvents = make([]EventReader, len(events))
|
||||
|
||||
es.interceptorMutex.Lock()
|
||||
defer es.interceptorMutex.Unlock()
|
||||
|
||||
for i, event := range events {
|
||||
interceptors, ok := es.eventInterceptors[EventType(event.Type)]
|
||||
if !ok || interceptors.eventMapper == nil {
|
||||
mappedEvents[i] = BaseEventFromRepo(event)
|
||||
//TODO: return error if unable to map event
|
||||
continue
|
||||
// return nil, errors.ThrowPreconditionFailed(nil, "V2-usujB", "event mapper not defined")
|
||||
}
|
||||
mappedEvents[i], err = interceptors.eventMapper(event)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return mappedEvents, nil
|
||||
}
|
||||
|
||||
type reducer interface {
|
||||
//Reduce handles the events of the internal events list
|
||||
// it only appends the newly added events
|
||||
Reduce() error
|
||||
//AppendEvents appends the passed events to an internal list of events
|
||||
AppendEvents(...EventReader)
|
||||
}
|
||||
|
||||
//FilterToReducer filters the events based on the search query, appends all events to the reducer and calls it's reduce function
|
||||
func (es *Eventstore) FilterToReducer(ctx context.Context, searchQuery *SearchQueryBuilder, r reducer) error {
|
||||
events, err := es.FilterEvents(ctx, searchQuery)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
go notify(aggregates)
|
||||
return nil
|
||||
r.AppendEvents(events...)
|
||||
|
||||
return r.Reduce()
|
||||
}
|
||||
|
||||
func (es *eventstore) FilterEvents(ctx context.Context, searchQuery *models.SearchQuery) ([]*models.Event, error) {
|
||||
if err := searchQuery.Validate(); err != nil {
|
||||
return nil, err
|
||||
//LatestSequence filters the latest sequence for the given search query
|
||||
func (es *Eventstore) LatestSequence(ctx context.Context, queryFactory *SearchQueryBuilder) (uint64, error) {
|
||||
query, err := queryFactory.build()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return es.repo.Filter(ctx, models.FactoryFromSearchQuery(searchQuery))
|
||||
return es.repo.LatestSequence(ctx, query)
|
||||
}
|
||||
|
||||
func (es *eventstore) LatestSequence(ctx context.Context, queryFactory *models.SearchQueryFactory) (uint64, error) {
|
||||
sequenceFactory := *queryFactory
|
||||
sequenceFactory = *(&sequenceFactory).Columns(models.Columns_Max_Sequence)
|
||||
sequenceFactory = *(&sequenceFactory).SequenceGreater(0)
|
||||
return es.repo.LatestSequence(ctx, &sequenceFactory)
|
||||
type queryReducer interface {
|
||||
reducer
|
||||
//Query returns the SearchQueryFactory for the events needed in reducer
|
||||
Query() *SearchQueryBuilder
|
||||
}
|
||||
|
||||
func (es *eventstore) Health(ctx context.Context) error {
|
||||
return es.repo.Health(ctx)
|
||||
//FilterToQueryReducer filters the events based on the search query of the query function,
|
||||
// appends all events to the reducer and calls it's reduce function
|
||||
func (es *Eventstore) FilterToQueryReducer(ctx context.Context, r queryReducer) error {
|
||||
events, err := es.FilterEvents(ctx, r.Query())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
r.AppendEvents(events...)
|
||||
|
||||
return r.Reduce()
|
||||
}
|
||||
|
||||
func (es *eventstore) V2() *es_v2.Eventstore {
|
||||
return es.esV2
|
||||
//RegisterFilterEventMapper registers a function for mapping an eventstore event to an event
|
||||
func (es *Eventstore) RegisterFilterEventMapper(eventType EventType, mapper func(*repository.Event) (EventReader, error)) *Eventstore {
|
||||
if mapper == nil || eventType == "" {
|
||||
return es
|
||||
}
|
||||
es.interceptorMutex.Lock()
|
||||
defer es.interceptorMutex.Unlock()
|
||||
|
||||
interceptor := es.eventInterceptors[eventType]
|
||||
interceptor.eventMapper = mapper
|
||||
es.eventInterceptors[eventType] = interceptor
|
||||
|
||||
return es
|
||||
}
|
||||
|
||||
func eventData(event EventPusher) ([]byte, error) {
|
||||
switch data := event.Data().(type) {
|
||||
case nil:
|
||||
return nil, nil
|
||||
case []byte:
|
||||
if json.Valid(data) {
|
||||
return data, nil
|
||||
}
|
||||
return nil, errors.ThrowInvalidArgument(nil, "V2-6SbbS", "data bytes are not json")
|
||||
}
|
||||
dataType := reflect.TypeOf(event.Data())
|
||||
if dataType.Kind() == reflect.Ptr {
|
||||
dataType = dataType.Elem()
|
||||
}
|
||||
if dataType.Kind() == reflect.Struct {
|
||||
dataBytes, err := json.Marshal(event.Data())
|
||||
if err != nil {
|
||||
return nil, errors.ThrowInvalidArgument(err, "V2-xG87M", "could not marshal data")
|
||||
}
|
||||
return dataBytes, nil
|
||||
}
|
||||
return nil, errors.ThrowInvalidArgument(nil, "V2-91NRm", "wrong type of event data")
|
||||
}
|
||||
|
||||
func uniqueConstraintActionToRepository(action UniqueConstraintAction) repository.UniqueConstraintAction {
|
||||
switch action {
|
||||
case UniqueConstraintAdd:
|
||||
return repository.UniqueConstraintAdd
|
||||
case UniqueConstraintRemove:
|
||||
return repository.UniqueConstraintRemoved
|
||||
default:
|
||||
return repository.UniqueConstraintAdd
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user