mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-13 19:44:21 +00:00
56b916a2b0
* begin init checks for projections * first projection checks * debug notification providers with query fixes * more projections and first index * more projections * more projections * finish projections * fix tests (remove db name) * create tables in setup * fix logging / error handling * add tenant to views * rename tenant to instance_id * add instance_id to all projections * add instance_id to all queries * correct instance_id on projections * add instance_id to failed_events * use separate context for instance * implement features projection * implement features projection * remove unique constraint from setup when migration failed * add error to failed setup event * add instance_id to primary keys * fix IAM projection * remove old migrations folder * fix keysFromYAML test
86 lines
2.2 KiB
Go
Executable File
86 lines
2.2 KiB
Go
Executable File
package query
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/caos/logging"
|
|
"github.com/getsentry/sentry-go"
|
|
|
|
v1 "github.com/caos/zitadel/internal/eventstore/v1"
|
|
"github.com/caos/zitadel/internal/eventstore/v1/models"
|
|
)
|
|
|
|
const (
|
|
eventLimit = 10000
|
|
)
|
|
|
|
type Handler interface {
|
|
ViewModel() string
|
|
EventQuery() (*models.SearchQuery, error)
|
|
Reduce(*models.Event) error
|
|
OnError(event *models.Event, err error) error
|
|
OnSuccess() error
|
|
MinimumCycleDuration() time.Duration
|
|
LockDuration() time.Duration
|
|
QueryLimit() uint64
|
|
|
|
AggregateTypes() []models.AggregateType
|
|
CurrentSequence() (uint64, error)
|
|
Eventstore() v1.Eventstore
|
|
|
|
Subscription() *v1.Subscription
|
|
}
|
|
|
|
func ReduceEvent(handler Handler, event *models.Event) {
|
|
defer func() {
|
|
err := recover()
|
|
|
|
if err != nil {
|
|
sentry.CurrentHub().Recover(err)
|
|
handler.Subscription().Unsubscribe()
|
|
}
|
|
}()
|
|
currentSequence, err := handler.CurrentSequence()
|
|
if err != nil {
|
|
logging.New().WithError(err).Warn("unable to get current sequence")
|
|
return
|
|
}
|
|
|
|
searchQuery := models.NewSearchQuery().
|
|
AggregateTypeFilter(handler.AggregateTypes()...).
|
|
SequenceBetween(currentSequence, event.Sequence).
|
|
SetLimit(eventLimit)
|
|
|
|
unprocessedEvents, err := handler.Eventstore().FilterEvents(context.Background(), searchQuery)
|
|
if err != nil {
|
|
logging.WithFields("HANDL-L6YH1", "sequence", event.Sequence).Warn("filter failed")
|
|
return
|
|
}
|
|
|
|
for _, unprocessedEvent := range unprocessedEvents {
|
|
currentSequence, err := handler.CurrentSequence()
|
|
if err != nil {
|
|
logging.Log("HANDL-BmpkC").WithError(err).Warn("unable to get current sequence")
|
|
return
|
|
}
|
|
if unprocessedEvent.Sequence < currentSequence {
|
|
logging.WithFields(
|
|
"unprocessed", unprocessedEvent.Sequence,
|
|
"current", currentSequence,
|
|
"view", handler.ViewModel()).
|
|
Warn("sequence not matching")
|
|
return
|
|
}
|
|
|
|
err = handler.Reduce(unprocessedEvent)
|
|
logging.WithFields("HANDL-V42TI", "sequence", unprocessedEvent.Sequence).OnError(err).Warn("reduce failed")
|
|
}
|
|
if len(unprocessedEvents) == eventLimit {
|
|
logging.WithFields("QUERY-BSqe9", "sequence", event.Sequence).Warn("didnt process event")
|
|
return
|
|
}
|
|
err = handler.Reduce(event)
|
|
logging.WithFields("HANDL-wQDL2", "sequence", event.Sequence).OnError(err).Warn("reduce failed")
|
|
}
|