diff --git a/cmd/setup/17.go b/cmd/setup/17.go new file mode 100644 index 000000000..9612f3d27 --- /dev/null +++ b/cmd/setup/17.go @@ -0,0 +1,26 @@ +package setup + +import ( + "context" + _ "embed" + + "github.com/zitadel/zitadel/internal/database" +) + +var ( + //go:embed 17.sql + addOffsetField string +) + +type AddOffsetToCurrentStates struct { + dbClient *database.DB +} + +func (mig *AddOffsetToCurrentStates) Execute(ctx context.Context) error { + _, err := mig.dbClient.ExecContext(ctx, addOffsetField) + return err +} + +func (mig *AddOffsetToCurrentStates) String() string { + return "17_add_offset_col_to_current_states" +} diff --git a/cmd/setup/17.sql b/cmd/setup/17.sql new file mode 100644 index 000000000..1c123bcb9 --- /dev/null +++ b/cmd/setup/17.sql @@ -0,0 +1 @@ +ALTER TABLE projections.current_states ADD filter_offset INTEGER; \ No newline at end of file diff --git a/cmd/setup/config.go b/cmd/setup/config.go index c2a2e9eda..668537eaf 100644 --- a/cmd/setup/config.go +++ b/cmd/setup/config.go @@ -61,19 +61,20 @@ func MustNewConfig(v *viper.Viper) *Config { } type Steps struct { - s1ProjectionTable *ProjectionTable - s2AssetsTable *AssetTable - FirstInstance *FirstInstance - s5LastFailed *LastFailed - s6OwnerRemoveColumns *OwnerRemoveColumns - s7LogstoreTables *LogstoreTables - s8AuthTokens *AuthTokenIndexes - CorrectCreationDate *CorrectCreationDate - s12AddOTPColumns *AddOTPColumns - s13FixQuotaProjection *FixQuotaConstraints - s14NewEventsTable *NewEventsTable - s15CurrentStates *CurrentProjectionState - s16UniqueConstraintsLower *UniqueConstraintToLower + s1ProjectionTable *ProjectionTable + s2AssetsTable *AssetTable + FirstInstance *FirstInstance + s5LastFailed *LastFailed + s6OwnerRemoveColumns *OwnerRemoveColumns + s7LogstoreTables *LogstoreTables + s8AuthTokens *AuthTokenIndexes + CorrectCreationDate *CorrectCreationDate + s12AddOTPColumns *AddOTPColumns + s13FixQuotaProjection *FixQuotaConstraints + s14NewEventsTable *NewEventsTable + s15CurrentStates *CurrentProjectionState + s16UniqueConstraintsLower *UniqueConstraintToLower + s17AddOffsetToUniqueConstraints *AddOffsetToCurrentStates } type encryptionKeyConfig struct { diff --git a/cmd/setup/setup.go b/cmd/setup/setup.go index 7b30340c5..a31c5e9ae 100644 --- a/cmd/setup/setup.go +++ b/cmd/setup/setup.go @@ -101,6 +101,7 @@ func Setup(config *Config, steps *Steps, masterKey string) { steps.s14NewEventsTable = &NewEventsTable{dbClient: esPusherDBClient} steps.s15CurrentStates = &CurrentProjectionState{dbClient: zitadelDBClient} steps.s16UniqueConstraintsLower = &UniqueConstraintToLower{dbClient: zitadelDBClient} + steps.s17AddOffsetToUniqueConstraints = &AddOffsetToCurrentStates{dbClient: zitadelDBClient} err = projection.Create(ctx, zitadelDBClient, eventstoreClient, config.Projections, nil, nil, nil) logging.OnError(err).Fatal("unable to start projections") @@ -143,6 +144,8 @@ func Setup(config *Config, steps *Steps, masterKey string) { logging.WithFields("name", steps.s15CurrentStates.String()).OnError(err).Fatal("migration failed") err = migration.Migrate(ctx, eventstoreClient, steps.s16UniqueConstraintsLower) logging.WithFields("name", steps.s16UniqueConstraintsLower.String()).OnError(err).Fatal("migration failed") + err = migration.Migrate(ctx, eventstoreClient, steps.s17AddOffsetToUniqueConstraints) + logging.WithFields("name", steps.s17AddOffsetToUniqueConstraints.String()).OnError(err).Fatal("migration failed") for _, repeatableStep := range repeatableSteps { err = migration.Migrate(ctx, eventstoreClient, repeatableStep) diff --git a/internal/command/preparation/command.go b/internal/command/preparation/command.go index 96a3da890..ce383cb8d 100644 --- a/internal/command/preparation/command.go +++ b/internal/command/preparation/command.go @@ -70,12 +70,9 @@ func transactionFilter(filter FilterToQueryReducer, commands []eventstore.Comman if err != nil { return nil, err } - for _, command := range commands { - event := command.(eventstore.Event) - if !query.Matches(event, len(events)) { - continue - } - events = append(events, event) + matches := query.Matches(commands...) + for _, command := range matches { + events = append(events, command.(eventstore.Event)) } return events, nil } diff --git a/internal/eventstore/aggregate.go b/internal/eventstore/aggregate.go index e6b90fd88..30053079d 100644 --- a/internal/eventstore/aggregate.go +++ b/internal/eventstore/aggregate.go @@ -81,7 +81,7 @@ type Aggregate struct { // AggregateType is the object name type AggregateType string -func isAggreagteTypes(a *Aggregate, types ...AggregateType) bool { +func isAggregateTypes(a *Aggregate, types ...AggregateType) bool { for _, typ := range types { if a.Type == typ { return true diff --git a/internal/eventstore/event.go b/internal/eventstore/event.go index 6469db5d6..4cb1d2ddc 100644 --- a/internal/eventstore/event.go +++ b/internal/eventstore/event.go @@ -97,9 +97,9 @@ func GenericEventMapper[T any, PT BaseEventSetter[T]](event Event) (Event, error return e, nil } -func isEventTypes(event Event, types ...EventType) bool { +func isEventTypes(command Command, types ...EventType) bool { for _, typ := range types { - if event.Type() == typ { + if command.Type() == typ { return true } } diff --git a/internal/eventstore/handler/v2/failed_event.go b/internal/eventstore/handler/v2/failed_event.go index 68771042e..b9630f7bc 100644 --- a/internal/eventstore/handler/v2/failed_event.go +++ b/internal/eventstore/handler/v2/failed_event.go @@ -47,7 +47,7 @@ func failureFromStatement(statement *Statement, err error) *failure { } } -func (h *Handler) handleFailedStmt(tx *sql.Tx, currentState *state, f *failure) (shouldContinue bool) { +func (h *Handler) handleFailedStmt(tx *sql.Tx, f *failure) (shouldContinue bool) { failureCount, err := h.failureCount(tx, f) if err != nil { h.logFailure(f).WithError(err).Warn("unable to get failure count") diff --git a/internal/eventstore/handler/v2/handler.go b/internal/eventstore/handler/v2/handler.go index afe277862..ada12c461 100644 --- a/internal/eventstore/handler/v2/handler.go +++ b/internal/eventstore/handler/v2/handler.go @@ -331,7 +331,11 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add var statements []*Statement statements, additionalIteration, err = h.generateStatements(ctx, tx, currentState) - if err != nil || len(statements) == 0 { + if err != nil { + return additionalIteration, err + } + if len(statements) == 0 { + err = h.setState(tx, currentState) return additionalIteration, err } @@ -341,6 +345,7 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add } currentState.position = statements[lastProcessedIndex].Position + currentState.offset = statements[lastProcessedIndex].offset currentState.aggregateID = statements[lastProcessedIndex].AggregateID currentState.aggregateType = statements[lastProcessedIndex].AggregateType currentState.sequence = statements[lastProcessedIndex].Sequence @@ -365,37 +370,44 @@ func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentSta return nil, false, err } eventAmount := len(events) - events = skipPreviouslyReduced(events, currentState) - - if len(events) == 0 { - h.updateLastUpdated(ctx, tx, currentState) - return nil, false, nil - } statements, err := h.eventsToStatements(tx, events, currentState) - if len(statements) == 0 { + if err != nil || len(statements) == 0 { return nil, false, err } + idx := skipPreviouslyReduced(statements, currentState) + if idx+1 == len(statements) { + currentState.position = statements[len(statements)-1].Position + currentState.offset = statements[len(statements)-1].offset + currentState.aggregateID = statements[len(statements)-1].AggregateID + currentState.aggregateType = statements[len(statements)-1].AggregateType + currentState.sequence = statements[len(statements)-1].Sequence + currentState.eventTimestamp = statements[len(statements)-1].CreationDate + + return nil, false, nil + } + statements = statements[idx+1:] + additionalIteration = eventAmount == int(h.bulkLimit) if len(statements) < len(events) { - // retry imediatly if statements failed + // retry immediately if statements failed additionalIteration = true } return statements, additionalIteration, nil } -func skipPreviouslyReduced(events []eventstore.Event, currentState *state) []eventstore.Event { - for i, event := range events { - if event.Position() == currentState.position && - event.Aggregate().ID == currentState.aggregateID && - event.Aggregate().Type == currentState.aggregateType && - event.Sequence() == currentState.sequence { - return events[i+1:] +func skipPreviouslyReduced(statements []*Statement, currentState *state) int { + for i, statement := range statements { + if statement.Position == currentState.position && + statement.AggregateID == currentState.aggregateID && + statement.AggregateType == currentState.aggregateType && + statement.Sequence == currentState.sequence { + return i } } - return events + return -1 } func (h *Handler) executeStatements(ctx context.Context, tx *sql.Tx, currentState *state, statements []*Statement) (lastProcessedIndex int, err error) { @@ -434,7 +446,7 @@ func (h *Handler) executeStatement(ctx context.Context, tx *sql.Tx, currentState if err = statement.Execute(tx, h.projection.Name()); err != nil { h.log().WithError(err).Error("statement execution failed") - shouldContinue = h.handleFailedStmt(tx, currentState, failureFromStatement(statement, err)) + shouldContinue = h.handleFailedStmt(tx, failureFromStatement(statement, err)) if shouldContinue { return nil } @@ -454,7 +466,11 @@ func (h *Handler) eventQuery(currentState *state) *eventstore.SearchQueryBuilder InstanceID(currentState.instanceID) if currentState.position > 0 { + // decrease position by 10 because builder.PositionAfter filters for position > and we need position >= builder = builder.PositionAfter(math.Float64frombits(math.Float64bits(currentState.position) - 10)) + if currentState.offset > 0 { + builder = builder.Offset(currentState.offset) + } } for aggregateType, eventTypes := range h.eventTypes { @@ -469,7 +485,7 @@ func (h *Handler) eventQuery(currentState *state) *eventstore.SearchQueryBuilder return builder } -// ProjectionName returns the name of the unlying projection. +// ProjectionName returns the name of the underlying projection. func (h *Handler) ProjectionName() string { return h.projection.Name() } diff --git a/internal/eventstore/handler/v2/state.go b/internal/eventstore/handler/v2/state.go index a5171be8f..4506135d7 100644 --- a/internal/eventstore/handler/v2/state.go +++ b/internal/eventstore/handler/v2/state.go @@ -19,6 +19,7 @@ type state struct { aggregateType eventstore.AggregateType aggregateID string sequence uint64 + offset uint16 } var ( @@ -30,8 +31,6 @@ var ( updateStateStmt string //go:embed state_lock.sql lockStateStmt string - //go:embed state_set_last_run.sql - updateStateLastRunStmt string errJustUpdated = errors.New("projection was just updated") ) @@ -47,6 +46,7 @@ func (h *Handler) currentState(ctx context.Context, tx *sql.Tx, config *triggerC sequence = new(sql.NullInt64) timestamp = new(sql.NullTime) position = new(sql.NullFloat64) + offset = new(sql.NullInt16) ) stateQuery := currentStateStmt @@ -61,6 +61,7 @@ func (h *Handler) currentState(ctx context.Context, tx *sql.Tx, config *triggerC sequence, timestamp, position, + offset, ) if errors.Is(err, sql.ErrNoRows) { err = h.lockState(tx, currentState.instanceID) @@ -75,6 +76,7 @@ func (h *Handler) currentState(ctx context.Context, tx *sql.Tx, config *triggerC currentState.sequence = uint64(sequence.Int64) currentState.eventTimestamp = timestamp.Time currentState.position = position.Float64 + currentState.offset = uint16(offset.Int16) return currentState, nil } @@ -87,6 +89,7 @@ func (h *Handler) setState(tx *sql.Tx, updatedState *state) error { updatedState.sequence, updatedState.eventTimestamp, updatedState.position, + updatedState.offset, ) if err != nil { h.log().WithError(err).Debug("unable to update state") @@ -99,11 +102,6 @@ func (h *Handler) setState(tx *sql.Tx, updatedState *state) error { return nil } -func (h *Handler) updateLastUpdated(ctx context.Context, tx *sql.Tx, updatedState *state) { - _, err := tx.ExecContext(ctx, updateStateLastRunStmt, h.projection.Name(), updatedState.instanceID) - h.log().OnError(err).Debug("unable to update last updated") -} - func (h *Handler) lockState(tx *sql.Tx, instanceID string) error { res, err := tx.Exec(lockStateStmt, h.projection.Name(), diff --git a/internal/eventstore/handler/v2/state_get.sql b/internal/eventstore/handler/v2/state_get.sql index 37e0c8784..a2082e484 100644 --- a/internal/eventstore/handler/v2/state_get.sql +++ b/internal/eventstore/handler/v2/state_get.sql @@ -4,6 +4,7 @@ SELECT , "sequence" , event_date , "position" + , filter_offset FROM projections.current_states WHERE diff --git a/internal/eventstore/handler/v2/state_get_await.sql b/internal/eventstore/handler/v2/state_get_await.sql index 0ab4bc06c..8c7587f7d 100644 --- a/internal/eventstore/handler/v2/state_get_await.sql +++ b/internal/eventstore/handler/v2/state_get_await.sql @@ -4,6 +4,7 @@ SELECT , "sequence" , event_date , "position" + , filter_offset FROM projections.current_states WHERE diff --git a/internal/eventstore/handler/v2/state_set.sql b/internal/eventstore/handler/v2/state_set.sql index 3a80bc200..2a79d80df 100644 --- a/internal/eventstore/handler/v2/state_set.sql +++ b/internal/eventstore/handler/v2/state_set.sql @@ -7,6 +7,7 @@ INSERT INTO projections.current_states ( , event_date , "position" , last_updated + , filter_offset ) VALUES ( $1 , $2 @@ -16,6 +17,7 @@ INSERT INTO projections.current_states ( , $6 , $7 , now() + , $8 ) ON CONFLICT ( projection_name , instance_id @@ -26,4 +28,5 @@ INSERT INTO projections.current_states ( , event_date = $6 , "position" = $7 , last_updated = statement_timestamp() + , filter_offset = $8 ; \ No newline at end of file diff --git a/internal/eventstore/handler/v2/state_set_last_run.sql b/internal/eventstore/handler/v2/state_set_last_run.sql deleted file mode 100644 index e632c47ba..000000000 --- a/internal/eventstore/handler/v2/state_set_last_run.sql +++ /dev/null @@ -1,2 +0,0 @@ -UPDATE projections.current_states SET last_updated = now() WHERE projection_name = $1 AND instance_id = $2; - \ No newline at end of file diff --git a/internal/eventstore/handler/v2/state_test.go b/internal/eventstore/handler/v2/state_test.go index 943219066..15031e66b 100644 --- a/internal/eventstore/handler/v2/state_test.go +++ b/internal/eventstore/handler/v2/state_test.go @@ -217,6 +217,7 @@ func TestHandler_updateLastUpdated(t *testing.T) { uint64(42), mock.AnyType[time.Time]{}, float64(42), + uint16(0), ), mock.WithExecRowsAffected(1), ), @@ -388,7 +389,7 @@ func TestHandler_currentState(t *testing.T) { "projection", ), mock.WithQueryResult( - []string{"aggregate_id", "aggregate_type", "event_sequence", "event_date", "position"}, + []string{"aggregate_id", "aggregate_type", "event_sequence", "event_date", "position", "offset"}, [][]driver.Value{ { "aggregate id", @@ -396,6 +397,7 @@ func TestHandler_currentState(t *testing.T) { int64(42), testTime, float64(42), + uint16(10), }, }, ), @@ -413,6 +415,7 @@ func TestHandler_currentState(t *testing.T) { aggregateType: "aggregate type", aggregateID: "aggregate id", sequence: 42, + offset: 10, }, }, }, diff --git a/internal/eventstore/handler/v2/statement.go b/internal/eventstore/handler/v2/statement.go index a167a3699..f2c915f63 100644 --- a/internal/eventstore/handler/v2/statement.go +++ b/internal/eventstore/handler/v2/statement.go @@ -17,15 +17,26 @@ import ( func (h *Handler) eventsToStatements(tx *sql.Tx, events []eventstore.Event, currentState *state) (statements []*Statement, err error) { statements = make([]*Statement, 0, len(events)) + + previousPosition := currentState.position + offset := currentState.offset for _, event := range events { statement, err := h.reduce(event) if err != nil { h.logEvent(event).WithError(err).Error("reduce failed") - if shouldContinue := h.handleFailedStmt(tx, currentState, failureFromEvent(event, err)); shouldContinue { + if shouldContinue := h.handleFailedStmt(tx, failureFromEvent(event, err)); shouldContinue { continue } return statements, err } + offset++ + if previousPosition != event.Position() { + // offset is 1 because we want to skip this event + offset = 1 + } + statement.offset = offset + statement.Position = event.Position() + previousPosition = event.Position() statements = append(statements, statement) } return statements, nil @@ -54,6 +65,8 @@ type Statement struct { CreationDate time.Time InstanceID string + offset uint16 + Execute Exec } diff --git a/internal/eventstore/repository/search_query.go b/internal/eventstore/repository/search_query.go index aa1068c1f..b0c7769cb 100644 --- a/internal/eventstore/repository/search_query.go +++ b/internal/eventstore/repository/search_query.go @@ -17,6 +17,7 @@ type SearchQuery struct { AllowTimeTravel bool AwaitOpenTransactions bool Limit uint64 + Offset uint16 Desc bool InstanceID *Filter @@ -121,12 +122,12 @@ func QueryFromBuilder(builder *eventstore.SearchQueryBuilder) (*SearchQuery, err query := &SearchQuery{ Columns: builder.GetColumns(), Limit: builder.GetLimit(), + Offset: builder.GetOffset(), Desc: builder.GetDesc(), Tx: builder.GetTx(), AllowTimeTravel: builder.GetAllowTimeTravel(), AwaitOpenTransactions: builder.GetAwaitOpenTransactions(), - // Queries: make([]*Filter, 0, 7), - SubQueries: make([][]*Filter, len(builder.GetQueries())), + SubQueries: make([][]*Filter, len(builder.GetQueries())), } for _, f := range []func(builder *eventstore.SearchQueryBuilder, query *SearchQuery) *Filter{ diff --git a/internal/eventstore/repository/sql/query.go b/internal/eventstore/repository/sql/query.go index 438ed9a13..a2eda79d3 100644 --- a/internal/eventstore/repository/sql/query.go +++ b/internal/eventstore/repository/sql/query.go @@ -89,6 +89,11 @@ func query(ctx context.Context, criteria querier, searchQuery *eventstore.Search query += " LIMIT ?" } + if q.Offset > 0 { + values = append(values, q.Offset) + query += " OFFSET ?" + } + query = criteria.placeholder(query) var contextQuerier interface { diff --git a/internal/eventstore/search_query.go b/internal/eventstore/search_query.go index 3f0212eb2..68321f9f9 100644 --- a/internal/eventstore/search_query.go +++ b/internal/eventstore/search_query.go @@ -14,6 +14,7 @@ import ( type SearchQueryBuilder struct { columns Columns limit uint64 + offset uint16 desc bool resourceOwner string instanceID *string @@ -37,6 +38,10 @@ func (b *SearchQueryBuilder) GetLimit() uint64 { return b.limit } +func (b *SearchQueryBuilder) GetOffset() uint16 { + return b.offset +} + func (b *SearchQueryBuilder) GetDesc() bool { return b.desc } @@ -149,25 +154,46 @@ func NewSearchQueryBuilder(columns Columns) *SearchQueryBuilder { } } -func (builder *SearchQueryBuilder) Matches(event Event, existingLen int) (matches bool) { - if builder.limit > 0 && uint64(existingLen) >= builder.limit { +func (builder *SearchQueryBuilder) Matches(commands ...Command) []Command { + matches := make([]Command, 0, len(commands)) + for i, command := range commands { + if builder.limit > 0 && builder.limit <= uint64(len(matches)) { + break + } + if builder.offset > 0 && uint16(i) < builder.offset { + continue + } + + if builder.matchCommand(command) { + matches = append(matches, command) + } + } + + return matches +} + +type sequencer interface { + Sequence() uint64 +} + +func (builder *SearchQueryBuilder) matchCommand(command Command) bool { + if builder.resourceOwner != "" && command.Aggregate().ResourceOwner != builder.resourceOwner { return false } - if builder.resourceOwner != "" && event.Aggregate().ResourceOwner != builder.resourceOwner { + if command.Aggregate().InstanceID != "" && builder.instanceID != nil && *builder.instanceID != "" && command.Aggregate().InstanceID != *builder.instanceID { return false } - if event.Aggregate().InstanceID != "" && builder.instanceID != nil && *builder.instanceID != "" && event.Aggregate().InstanceID != *builder.instanceID { - return false - } - if builder.eventSequenceGreater > 0 && event.Sequence() <= builder.eventSequenceGreater { - return false + if seq, ok := command.(sequencer); ok { + if builder.eventSequenceGreater > 0 && seq.Sequence() <= builder.eventSequenceGreater { + return false + } } if len(builder.queries) == 0 { return true } for _, query := range builder.queries { - if query.matches(event) { + if query.matches(command) { return true } } @@ -186,6 +212,12 @@ func (builder *SearchQueryBuilder) Limit(limit uint64) *SearchQueryBuilder { return builder } +// Limit defines how many events are returned maximally. +func (builder *SearchQueryBuilder) Offset(offset uint16) *SearchQueryBuilder { + builder.offset = offset + return builder +} + // ResourceOwner defines the resource owner (org) of the events func (builder *SearchQueryBuilder) ResourceOwner(resourceOwner string) *SearchQueryBuilder { builder.resourceOwner = resourceOwner @@ -317,14 +349,14 @@ func (query *SearchQuery) Builder() *SearchQueryBuilder { return query.builder } -func (query *SearchQuery) matches(event Event) bool { - if ok := isAggreagteTypes(event.Aggregate(), query.aggregateTypes...); len(query.aggregateTypes) > 0 && !ok { +func (query *SearchQuery) matches(command Command) bool { + if ok := isAggregateTypes(command.Aggregate(), query.aggregateTypes...); len(query.aggregateTypes) > 0 && !ok { return false } - if ok := isAggregateIDs(event.Aggregate(), query.aggregateIDs...); len(query.aggregateIDs) > 0 && !ok { + if ok := isAggregateIDs(command.Aggregate(), query.aggregateIDs...); len(query.aggregateIDs) > 0 && !ok { return false } - if ok := isEventTypes(event, query.eventTypes...); len(query.eventTypes) > 0 && !ok { + if ok := isEventTypes(command, query.eventTypes...); len(query.eventTypes) > 0 && !ok { return false } return true diff --git a/internal/eventstore/search_query_test.go b/internal/eventstore/search_query_test.go index a125c515b..8c654911e 100644 --- a/internal/eventstore/search_query_test.go +++ b/internal/eventstore/search_query_test.go @@ -251,20 +251,22 @@ func assertQuery(t *testing.T, i int, want, got *SearchQuery) { func TestSearchQuery_matches(t *testing.T) { type args struct { - event Event + event Command } tests := []struct { name string query *SearchQuery - event Event + event Command want bool }{ { name: "wrong aggregate type", query: NewSearchQueryBuilder(ColumnsEvent).AddQuery().AggregateTypes("searched"), - event: &BaseEvent{ - Agg: &Aggregate{ - Type: "found", + event: &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + Type: "found", + }, }, }, want: false, @@ -272,9 +274,11 @@ func TestSearchQuery_matches(t *testing.T) { { name: "wrong aggregate id", query: NewSearchQueryBuilder(ColumnsEvent).AddQuery().AggregateIDs("1", "10", "100"), - event: &BaseEvent{ - Agg: &Aggregate{ - ID: "2", + event: &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ID: "2", + }, }, }, want: false, @@ -282,9 +286,11 @@ func TestSearchQuery_matches(t *testing.T) { { name: "wrong event type", query: NewSearchQueryBuilder(ColumnsEvent).AddQuery().EventTypes("event.searched.type"), - event: &BaseEvent{ - EventType: "event.actual.type", - Agg: &Aggregate{}, + event: &matcherCommand{ + BaseEvent{ + EventType: "event.actual.type", + Agg: &Aggregate{}, + }, }, want: false, }, @@ -295,26 +301,30 @@ func TestSearchQuery_matches(t *testing.T) { AggregateIDs("2"). AggregateTypes("actual"). EventTypes("event.actual.type"), - event: &BaseEvent{ - Seq: 55, - Agg: &Aggregate{ - ID: "2", - Type: "actual", + event: &matcherCommand{ + BaseEvent{ + Seq: 55, + Agg: &Aggregate{ + ID: "2", + Type: "actual", + }, + EventType: "event.actual.type", }, - EventType: "event.actual.type", }, want: true, }, { name: "matching empty query", query: NewSearchQueryBuilder(ColumnsEvent).AddQuery(), - event: &BaseEvent{ - Seq: 55, - Agg: &Aggregate{ - ID: "2", - Type: "actual", + event: &matcherCommand{ + BaseEvent{ + Seq: 55, + Agg: &Aggregate{ + ID: "2", + Type: "actual", + }, + EventType: "event.actual.type", }, - EventType: "event.actual.type", }, want: true, }, @@ -334,76 +344,128 @@ func TestSearchQuery_matches(t *testing.T) { } } +type matcherCommand struct { + BaseEvent +} + +func (matcherCommand) Payload() any { return nil } + +func (matcherCommand) UniqueConstraints() []*UniqueConstraint { return nil } + func TestSearchQueryBuilder_Matches(t *testing.T) { type args struct { - event Event - existingLen int + commands []Command } tests := []struct { - name string - builder *SearchQueryBuilder - args args - want bool + name string + builder *SearchQueryBuilder + args args + wantedLen int }{ { - name: "sequence too high", - builder: NewSearchQueryBuilder(ColumnsEvent).SequenceGreater(60), + name: "sequence too high", + builder: NewSearchQueryBuilder(ColumnsEvent). + SequenceGreater(60), args: args{ - event: &BaseEvent{ - Agg: &Aggregate{ - InstanceID: "instance", - }, - Seq: 60, - }, - }, - want: false, - }, - { - name: "limit exeeded", - builder: NewSearchQueryBuilder(ColumnsEvent).Limit(100), - args: args{ - event: &BaseEvent{}, - existingLen: 100, - }, - want: false, - }, - { - name: "wrong resource owner", - builder: NewSearchQueryBuilder(ColumnsEvent).ResourceOwner("query"), - args: args{ - event: &BaseEvent{ - Agg: &Aggregate{ - ResourceOwner: "ro", + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + }, + Seq: 60, + }, }, }, - existingLen: 0, }, - want: false, + wantedLen: 0, }, { - name: "wrong instance", - builder: NewSearchQueryBuilder(ColumnsEvent).InstanceID("instance"), + name: "limit exeeded", + builder: NewSearchQueryBuilder(ColumnsEvent). + Limit(2), args: args{ - event: &BaseEvent{ - Agg: &Aggregate{ - InstanceID: "different instance", + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro", + InstanceID: "instance", + }, + Seq: 1001, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro", + InstanceID: "instance", + }, + Seq: 1001, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro", + InstanceID: "instance", + }, + Seq: 1001, + }, }, }, - existingLen: 0, }, - want: false, + wantedLen: 2, }, { - name: "query failed", - builder: NewSearchQueryBuilder(ColumnsEvent).SequenceGreater(1000), + name: "wrong resource owner", + builder: NewSearchQueryBuilder(ColumnsEvent). + ResourceOwner("query"), args: args{ - event: &BaseEvent{ - Seq: 999, - Agg: &Aggregate{}, + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro", + }, + }, + }, }, - existingLen: 0, }, - want: false, + wantedLen: 0, + }, + { + name: "wrong instance", + builder: NewSearchQueryBuilder(ColumnsEvent). + InstanceID("instance"), + args: args{ + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "different instance", + }, + }, + }, + }, + }, + wantedLen: 0, + }, + { + name: "query failed", + builder: NewSearchQueryBuilder(ColumnsEvent). + SequenceGreater(1000), + args: args{ + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Seq: 999, + Agg: &Aggregate{}, + }, + }, + }, + }, + wantedLen: 0, }, { name: "matching", @@ -413,65 +475,242 @@ func TestSearchQueryBuilder_Matches(t *testing.T) { InstanceID("instance"). SequenceGreater(1000), args: args{ - event: &BaseEvent{ - Agg: &Aggregate{ - ResourceOwner: "ro", - InstanceID: "instance", + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro", + InstanceID: "instance", + }, + Seq: 1001, + }, }, - Seq: 1001, }, - existingLen: 999, }, - want: true, + wantedLen: 1, }, { - name: "matching builder resourceOwner and Instance", - builder: NewSearchQueryBuilder(ColumnsEvent), + name: "matching builder resourceOwner and Instance", + builder: NewSearchQueryBuilder(ColumnsEvent). + ResourceOwner("ro"). + InstanceID("instance"), args: args{ - event: &BaseEvent{ - Agg: &Aggregate{ - ResourceOwner: "ro", - InstanceID: "instance", + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro", + InstanceID: "instance", + }, + Seq: 1001, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro2", + InstanceID: "instance2", + }, + Seq: 1002, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro2", + InstanceID: "instance", + }, + Seq: 1003, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro", + InstanceID: "instance2", + }, + Seq: 1004, + }, }, - Seq: 1001, }, - existingLen: 999, }, - want: true, + wantedLen: 1, }, { - name: "matching builder resourceOwner only", - builder: NewSearchQueryBuilder(ColumnsEvent), + name: "matching builder resourceOwner only", + builder: NewSearchQueryBuilder(ColumnsEvent). + ResourceOwner("ro"), args: args{ - event: &BaseEvent{ - Agg: &Aggregate{ - ResourceOwner: "ro", + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro", + }, + Seq: 1001, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + ResourceOwner: "ro2", + }, + Seq: 1001, + }, }, - Seq: 1001, }, - existingLen: 999, }, - want: true, + wantedLen: 1, }, { - name: "matching builder instanceID only", - builder: NewSearchQueryBuilder(ColumnsEvent), + name: "matching builder instanceID only", + builder: NewSearchQueryBuilder(ColumnsEvent). + InstanceID("instance"), args: args{ - event: &BaseEvent{ - Agg: &Aggregate{ - InstanceID: "instance", + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + }, + Seq: 1001, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance2", + }, + Seq: 1001, + }, }, - Seq: 1001, }, - existingLen: 999, }, - want: true, + wantedLen: 1, + }, + { + name: "offset too high", + builder: NewSearchQueryBuilder(ColumnsEvent). + Offset(2), + args: args{ + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + }, + Seq: 1001, + }, + }, + }, + }, + wantedLen: 0, + }, + { + name: "offset", + builder: NewSearchQueryBuilder(ColumnsEvent). + Offset(1), + args: args{ + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + }, + Seq: 1001, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + }, + Seq: 1002, + }, + }, + }, + }, + wantedLen: 1, + }, + { + name: "offset and limit", + builder: NewSearchQueryBuilder(ColumnsEvent). + Offset(1). + Limit(1), + args: args{ + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + }, + Seq: 1001, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + }, + Seq: 1002, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + }, + Seq: 1002, + }, + }, + }, + }, + wantedLen: 1, + }, + { + name: "sub query", + builder: NewSearchQueryBuilder(ColumnsEvent). + AddQuery(). + AggregateTypes("test"). + Builder(), + args: args{ + commands: []Command{ + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + Type: "test", + }, + Seq: 1001, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + Type: "test", + }, + Seq: 1002, + }, + }, + &matcherCommand{ + BaseEvent{ + Agg: &Aggregate{ + InstanceID: "instance", + Type: "test2", + }, + Seq: 1003, + }, + }, + }, + }, + wantedLen: 2, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := tt.builder.Matches(tt.args.event, tt.args.existingLen); got != tt.want { - t.Errorf("SearchQueryBuilder.Matches() = %v, want %v", got, tt.want) + if got := tt.builder.Matches(tt.args.commands...); len(got) != tt.wantedLen { + t.Errorf("SearchQueryBuilder.Matches() = %v, wantted len %v", got, tt.wantedLen) } }) }