mirror of
https://github.com/zitadel/zitadel.git
synced 2025-08-12 03:27:32 +00:00
feat(queries): use org projection (#2342)
* job queue * wg improvements * start handler * statement * statements * imporve handler * improve statement * statement in seperate file * move handlers * move query/old to query * handler * read models * bulk works * cleanup * contrib * rename readmodel to projection * rename read_models schema to projections * rename read_models schema to projections * search query as func, bulk iterates as long as new events * add event sequence less query * update checks for events between current sequence and sequence of first statement if it has previous sequence 0 * cleanup crdb projection * refactor projection handler * start with testing * tests for handler * remove todo * refactor statement: remove table name, add tests * improve projection handler shutdown, no savepoint if noop stmt, tests for stmt handler * tests * start failed events * seperate branch for contrib * move statement constructors to crdb pkg * correct import * Subscribe for eventtypes (#1800) * fix: is default (#1737) * fix: use email as username on global org (#1738) * fix: use email as username on global org * Update user_human.go * Update register_handler.go * chore(deps): update docusaurus (#1739) * chore: remove PAT and use GH Token (#1716) * chore: remove PAT and use GH Token * fix env * fix env * fix env * md lint * trigger ci * change user * fix GH bug * replace login part * chore: add GH Token to sem rel (#1746) * chore: add GH Token to sem rel * try branch * add GH Token * remove test branch again * docs: changes acme to acme-caos (#1744) * changes acme to acme-caos * Apply suggestions from code review Co-authored-by: Florian Forster <florian@caos.ch> Co-authored-by: Maximilian Panne <maximilian.panne@gmail.com> Co-authored-by: Florian Forster <florian@caos.ch> * feat: add additional origins on applications (#1691) * feat: add additional origins on applications * app additional redirects * chore(deps-dev): bump @angular/cli from 11.2.8 to 11.2.11 in /console (#1706) * fix: show org with regex (#1688) * fix: flag mapping (#1699) * chore(deps-dev): bump @angular/cli from 11.2.8 to 11.2.11 in /console Bumps [@angular/cli](https://github.com/angular/angular-cli) from 11.2.8 to 11.2.11. - [Release notes](https://github.com/angular/angular-cli/releases) - [Commits](https://github.com/angular/angular-cli/compare/v11.2.8...v11.2.11) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Silvan <silvan.reusser@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * chore(deps-dev): bump stylelint from 13.10.0 to 13.13.1 in /console (#1703) * fix: show org with regex (#1688) * fix: flag mapping (#1699) * chore(deps-dev): bump stylelint from 13.10.0 to 13.13.1 in /console Bumps [stylelint](https://github.com/stylelint/stylelint) from 13.10.0 to 13.13.1. - [Release notes](https://github.com/stylelint/stylelint/releases) - [Changelog](https://github.com/stylelint/stylelint/blob/master/CHANGELOG.md) - [Commits](https://github.com/stylelint/stylelint/compare/13.10.0...13.13.1) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Silvan <silvan.reusser@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * chore(deps-dev): bump @types/node from 14.14.37 to 15.0.1 in /console (#1702) * fix: show org with regex (#1688) * fix: flag mapping (#1699) * chore(deps-dev): bump @types/node from 14.14.37 to 15.0.1 in /console Bumps [@types/node](https://github.com/DefinitelyTyped/DefinitelyTyped/tree/HEAD/types/node) from 14.14.37 to 15.0.1. - [Release notes](https://github.com/DefinitelyTyped/DefinitelyTyped/releases) - [Commits](https://github.com/DefinitelyTyped/DefinitelyTyped/commits/HEAD/types/node) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Silvan <silvan.reusser@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * chore(deps): bump ts-protoc-gen from 0.14.0 to 0.15.0 in /console (#1701) * fix: show org with regex (#1688) * fix: flag mapping (#1699) * chore(deps): bump ts-protoc-gen from 0.14.0 to 0.15.0 in /console Bumps [ts-protoc-gen](https://github.com/improbable-eng/ts-protoc-gen) from 0.14.0 to 0.15.0. - [Release notes](https://github.com/improbable-eng/ts-protoc-gen/releases) - [Changelog](https://github.com/improbable-eng/ts-protoc-gen/blob/master/CHANGELOG.md) - [Commits](https://github.com/improbable-eng/ts-protoc-gen/compare/0.14.0...0.15.0) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Silvan <silvan.reusser@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * chore(deps-dev): bump @types/jasmine from 3.6.9 to 3.6.10 in /console (#1682) Bumps [@types/jasmine](https://github.com/DefinitelyTyped/DefinitelyTyped/tree/HEAD/types/jasmine) from 3.6.9 to 3.6.10. - [Release notes](https://github.com/DefinitelyTyped/DefinitelyTyped/releases) - [Commits](https://github.com/DefinitelyTyped/DefinitelyTyped/commits/HEAD/types/jasmine) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * chore(deps): bump @types/google-protobuf in /console (#1681) Bumps [@types/google-protobuf](https://github.com/DefinitelyTyped/DefinitelyTyped/tree/HEAD/types/google-protobuf) from 3.7.4 to 3.15.2. - [Release notes](https://github.com/DefinitelyTyped/DefinitelyTyped/releases) - [Commits](https://github.com/DefinitelyTyped/DefinitelyTyped/commits/HEAD/types/google-protobuf) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * chore(deps): bump grpc from 1.24.5 to 1.24.7 in /console (#1666) Bumps [grpc](https://github.com/grpc/grpc-node) from 1.24.5 to 1.24.7. - [Release notes](https://github.com/grpc/grpc-node/releases) - [Commits](https://github.com/grpc/grpc-node/compare/grpc@1.24.5...grpc@1.24.7) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * lock * chore(deps-dev): bump @angular/language-service from 11.2.9 to 11.2.12 in /console (#1704) * fix: show org with regex (#1688) * fix: flag mapping (#1699) * chore(deps-dev): bump @angular/language-service in /console Bumps [@angular/language-service](https://github.com/angular/angular/tree/HEAD/packages/language-service) from 11.2.9 to 11.2.12. - [Release notes](https://github.com/angular/angular/releases) - [Changelog](https://github.com/angular/angular/blob/master/CHANGELOG.md) - [Commits](https://github.com/angular/angular/commits/11.2.12/packages/language-service) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Silvan <silvan.reusser@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * package lock * downgrade grpc * downgrade protobuf types * revert npm packs 🥸 Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Silvan <silvan.reusser@gmail.com> * docs: update run and start section texts (#1745) * update run and start section texts * adds showcase Co-authored-by: Maximilian Panne <maximilian.panne@gmail.com> * fix: additional origin list (#1753) * fix: handle api configs in authz handler (#1755) * fix(console): add model for api keys, fix toast, binding (#1757) * fix: add model for api keys, fix toast, binding * show api clientid * fix: missing patchvalue (#1758) * feat: refresh token (#1728) * begin refresh tokens * refresh tokens * list and revoke refresh tokens * handle remove * tests for refresh tokens * uniqueness and default expiration * rename oidc token methods * cleanup * migration version * Update internal/static/i18n/en.yaml Co-authored-by: Fabi <38692350+fgerschwiler@users.noreply.github.com> * fixes * feat: update oidc pkg for refresh tokens Co-authored-by: Fabi <38692350+fgerschwiler@users.noreply.github.com> * fix: correct json name of clientId in key.json (#1760) * fix: migration version (#1767) * start subscription * eventtypes * fix(login): links (#1778) * fix(login): href for help * fix(login): correct link to tos * fix: access tokens for service users and refresh token infos (#1779) * fix: access token for service user * handle info from refresh request * uniqueness * postpone access token uniqueness change * chore(coc): recommend code of conduct (#1782) * subscribe for events * feat(console): refresh toggle out of granttype context (#1785) * refresh toggle * disable if not code flow, lint * lint * fix: change oidc config order * accept refresh option within flow Co-authored-by: Livio Amstutz <livio.a@gmail.com> * fix: refresh token activation (#1795) * fix: oidc grant type check * docs: add offline_access scope * docs: update refresh token status in supported grant types * fix: update oidc pkg * fix: check refresh token grant type (#1796) * configuration structs * org admins * failed events * fixes Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Livio Amstutz <livio.a@gmail.com> Co-authored-by: Florian Forster <florian@caos.ch> Co-authored-by: mffap <mpa@caos.ch> Co-authored-by: Maximilian Panne <maximilian.panne@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Fabi <38692350+fgerschwiler@users.noreply.github.com> * remove comment * aggregate reducer * remove eventtypes * add protoc-get-validate to mod * fix transaltion * upsert * add gender on org admins, allow to retry failed stmts after configurable time * remove if * sub queries * fix: tests * add builder to tests * new search query * rename searchquerybuilder to builder * remove comment from code * test with multiple queries * add filters test * current sequences * make org and org_admins work again * add aggregate type to current sequence * fix(contibute): listing * add validate module * fix: search queries * feat(eventstore): previous aggregate root sequence (#1810) * feat(eventstore): previous aggregate root sequence * fix tests * fix: eventstore v1 test * add col to all mocked rows * next try * fix mig * rename aggregate root to aggregate type * update comment Co-authored-by: Livio Amstutz <livio.a@gmail.com> Co-authored-by: Livio Amstutz <livio.a@gmail.com> * small refactorings * allow update multiple current sequences * unique log id * fix migrations * rename org admin to org owner * improve error handling and logging * fix(migration): optimize prev agg root seq * fix: projection handler test * fix: sub queries * small fixes * additional event types * correct org owner projection * fix primary key * feat(eventstore): jobs for projections (#2026) * fix: template names in login (#1974) * fix: template names in login * fix: error.html * fix: check for features on mgmt only (#1976) * fix: add sentry in ui, http and projection handlers (#1977) * fix: add sentry in ui, http and projection handlers * fix test * fix(eventstore): sub queries (#1805) * sub queries * fix: tests * add builder to tests * new search query * rename searchquerybuilder to builder * remove comment from code * test with multiple queries * add filters test * fix(contibute): listing * add validate module * fix: search queries * remove unused event type in query * ignore query if error in marshal * go mod tidy * update privacy policy query * update queries Co-authored-by: Livio Amstutz <livio.a@gmail.com> * feat: Extend oidc idp with oauth endpoints (#1980) * feat: add oauth attributes to oidc idp configuration * feat: return idpconfig id on create idp * feat: tests * feat: descriptions * feat: docs * feat: tests * docs: update to beta 3 (#1984) * fix: role assertion (#1986) * fix: enum to display access token role assertion * improve assertion descriptions * fix nil pointer * docs: eventstore (#1982) * docs: eventstore * Apply suggestions from code review Co-authored-by: Florian Forster <florian@caos.ch> Co-authored-by: Florian Forster <florian@caos.ch> * fix(sentry): trigger sentry release (#1989) * feat(send sentry release): send sentry release * fix(moved step and added releasetag): moved step and added releasetag * fix: set version for sentry release (#1990) * feat(send sentry release): send sentry release * fix(moved step and added releasetag): moved step and added releasetag * fix(corrected var name): corrected var name Co-authored-by: Livio Amstutz <livio.a@gmail.com> * fix: log error reason on terminate session (#1973) * fix: return default language file, if requested lang does not exist for default login texts (#1988) * fix: return default language file, if requested lang doesnt exists * feat: read default translation file * feat: docs * fix: race condition in auth request unmarshalling (#1993) * feat: handle ui_locales in login (#1994) * fix: handle ui_locales in login * move supportedlanguage func into i18n package * update oidc pkg * fix: handle closed channels on unsubscribe (#1995) * fix: give restore more time (#1997) * fix: translation file read (#2009) * feat: translation file read * feat: readme * fix: enable idp add button for iam users (#2010) * fix: filter event_data (#2011) * feat: Custom message files (#1992) * feat: add get custom message text to admin api * feat: read custom message texts from files * feat: get languages in apis * feat: get languages in apis * feat: get languages in apis * feat: pr feedback * feat: docs * feat: merge main * fix: sms notification (#2013) * fix: phone verifications * feat: fix password reset as sms * fix: phone verification * fix: grpc status in sentry and validation interceptors (#2012) * fix: remove oauth endpoints from oidc config proto (#2014) * try with view * fix(console): disable sw (#2021) * fix: disable sw * angular.json disable sw * project projections * fix typos * customize projections * customizable projections, add change date to projects Co-authored-by: Livio Amstutz <livio.a@gmail.com> Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Fabi <38692350+fgerschwiler@users.noreply.github.com> Co-authored-by: Florian Forster <florian@caos.ch> Co-authored-by: mffap <mpa@caos.ch> Co-authored-by: Christian Jakob <47860090+thesephirot@users.noreply.github.com> Co-authored-by: Elio Bischof <eliobischof@gmail.com> * env file * typo * correct users * correct migration * fix: merge fail * fix test * fix(tests): unordered matcher * improve currentSequenceMatcher * correct certs * correct certs * add zitadel database on database list * refctor switch in match * enable all handlers * Delete io.env * cleanup * add handlers * rename view to projection * rename view to projection * fix type typo * remove unnecessary logs * refactor stmts * simplify interval calculation * fix tests * fix unlock test * fix migration * migs * fix(operator): update cockroach and flyway versions (#2138) * chore(deps): bump k8s.io/apiextensions-apiserver from 0.19.2 to 0.21.3 Bumps [k8s.io/apiextensions-apiserver](https://github.com/kubernetes/apiextensions-apiserver) from 0.19.2 to 0.21.3. - [Release notes](https://github.com/kubernetes/apiextensions-apiserver/releases) - [Commits](https://github.com/kubernetes/apiextensions-apiserver/compare/v0.19.2...v0.21.3) --- updated-dependencies: - dependency-name: k8s.io/apiextensions-apiserver dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * chore(deps): bump google.golang.org/api from 0.34.0 to 0.52.0 Bumps [google.golang.org/api](https://github.com/googleapis/google-api-go-client) from 0.34.0 to 0.52.0. - [Release notes](https://github.com/googleapis/google-api-go-client/releases) - [Changelog](https://github.com/googleapis/google-api-go-client/blob/master/CHANGES.md) - [Commits](https://github.com/googleapis/google-api-go-client/compare/v0.34.0...v0.52.0) --- updated-dependencies: - dependency-name: google.golang.org/api dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * start update dependencies * update mods and otlp * fix(build): update to go 1.16 * old version for k8s mods * update k8s versions * update orbos * fix(operator): update cockroach and flyway version * Update images.go Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Stefan Benz <stefan@caos.ch> * fix import * fix typo * fix(migration): add org projection * fix(projection): correct table for org events in org owners * better insert stmt * fix typo * fix typo * set max connection lifetime * set max conns and conn lifetime in eventstore v1 * configure sql connection settings * add mig for agg type index * fix replace tab in yaml * handler interfaces * subscription * first try * handler * move sql client initialization * first part implemented * removed all occurencies of org by id and search orgs * fix merge issues * cleanup code * fix: queries implements orgviewprovider * cleanup * refactor text comparison * remove unused file * remove unused code * log * remove unused code * remove unused field * remove unused file * refactor * tests for search query * remove try * simplify state change mappers * projection tests * query functions * move reusable objects to separate files * rename domain column to primar_domain * fix tests * add current sequence * remove log prints * fix tests * fix: verifier * fix test * rename domain col migrations * simplify search response * add custom column constructors * fix: org projection table const * fix: full column name * feat: text query extension * fix: tests for query * number query * add deprection message * column in a single place (#2416) * column in a single place * use projection for columns * query column with aliases * rename methods * remove unused code * column for current sequences * global counter column * fix is org unique * fix: merge main and change actions / flow projections to new query side (#2434) * feat: actions (#2377) * feat(actions): begin api * feat(actions): begin api * api and projections * fix: handle multiple statements for a single event in projections * export func type * fix test * update to new reduce interface * flows in login * feat: jwt idp * feat: command side * feat: add tests * actions and flows * fill idp views with jwt idps and return apis * add jwtEndpoint to jwt idp * begin jwt request handling * add feature * merge * merge * handle jwt idp * cleanup * bug fixes * autoregister * get token from specific header name * fix: proto * fixes * i18n * begin tests * fix and log http proxy * remove docker cache * fixes * usergrants in actions api * tests adn cleanup * cleanup * fix add user grant * set login context * i18n Co-authored-by: fabi <fabienne.gerschwiler@gmail.com> * change actions / flow projections to new query side * fixes * enable org projection Co-authored-by: fabi <fabienne.gerschwiler@gmail.com> * fixes * cleanup * add tests Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Livio Amstutz <livio.a@gmail.com> Co-authored-by: Florian Forster <florian@caos.ch> Co-authored-by: mffap <mpa@caos.ch> Co-authored-by: Maximilian Panne <maximilian.panne@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Fabi <38692350+fgerschwiler@users.noreply.github.com> Co-authored-by: Christian Jakob <47860090+thesephirot@users.noreply.github.com> Co-authored-by: Elio Bischof <eliobischof@gmail.com> Co-authored-by: Stefan Benz <stefan@caos.ch> Co-authored-by: fabi <fabienne.gerschwiler@gmail.com>
This commit is contained in:
Silvan
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
|
||||
"github.com/caos/logging"
|
||||
|
||||
"github.com/caos/zitadel/internal/domain"
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore"
|
||||
@@ -12,13 +13,27 @@ import (
|
||||
"github.com/caos/zitadel/internal/repository/action"
|
||||
)
|
||||
|
||||
const (
|
||||
ActionTable = "zitadel.projections.actions"
|
||||
ActionIDCol = "id"
|
||||
ActionCreationDateCol = "creation_date"
|
||||
ActionChangeDateCol = "change_date"
|
||||
ActionResourceOwnerCol = "resource_owner"
|
||||
ActionStateCol = "action_state"
|
||||
ActionSequenceCol = "sequence"
|
||||
ActionNameCol = "name"
|
||||
ActionScriptCol = "script"
|
||||
ActionTimeoutCol = "timeout"
|
||||
ActionAllowedToFailCol = "allowed_to_fail"
|
||||
)
|
||||
|
||||
type ActionProjection struct {
|
||||
crdb.StatementHandler
|
||||
}
|
||||
|
||||
func NewActionProjection(ctx context.Context, config crdb.StatementHandlerConfig) *ActionProjection {
|
||||
p := &ActionProjection{}
|
||||
config.ProjectionName = "projections.actions"
|
||||
config.ProjectionName = ActionTable
|
||||
config.Reducers = p.reducers()
|
||||
p.StatementHandler = crdb.NewStatementHandler(ctx, config)
|
||||
return p
|
||||
@@ -54,38 +69,25 @@ func (p *ActionProjection) reducers() []handler.AggregateReducer {
|
||||
}
|
||||
}
|
||||
|
||||
const (
|
||||
actionIDCol = "id"
|
||||
actionCreationDateCol = "creation_date"
|
||||
actionChangeDateCol = "change_date"
|
||||
actionResourceOwnerCol = "resource_owner"
|
||||
actionStateCol = "action_state"
|
||||
actionSequenceCol = "sequence"
|
||||
actionNameCol = "name"
|
||||
actionScriptCol = "script"
|
||||
actionTimeoutCol = "timeout"
|
||||
actionAllowedToFailCol = "allowed_to_fail"
|
||||
)
|
||||
|
||||
func (p *ActionProjection) reduceActionAdded(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*action.AddedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-zWCk3", "seq", event.Sequence, "expectedType", action.AddedEventType).Error("was not an event")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-uYq4r", "reduce.wrong.event.type")
|
||||
logging.LogWithFields("HANDL-Sgg31", "seq", event.Sequence, "expectedType", action.AddedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-Dff21", "reduce.wrong.event.type")
|
||||
}
|
||||
return crdb.NewCreateStatement(
|
||||
e,
|
||||
[]handler.Column{
|
||||
handler.NewCol(actionIDCol, e.Aggregate().ID),
|
||||
handler.NewCol(actionCreationDateCol, e.CreationDate()),
|
||||
handler.NewCol(actionChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(actionResourceOwnerCol, e.Aggregate().ResourceOwner),
|
||||
handler.NewCol(actionSequenceCol, e.Sequence()),
|
||||
handler.NewCol(actionNameCol, e.Name),
|
||||
handler.NewCol(actionScriptCol, e.Script),
|
||||
handler.NewCol(actionTimeoutCol, e.Timeout),
|
||||
handler.NewCol(actionAllowedToFailCol, e.AllowedToFail),
|
||||
handler.NewCol(actionStateCol, domain.ActionStateActive),
|
||||
handler.NewCol(ActionIDCol, e.Aggregate().ID),
|
||||
handler.NewCol(ActionCreationDateCol, e.CreationDate()),
|
||||
handler.NewCol(ActionChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(ActionResourceOwnerCol, e.Aggregate().ResourceOwner),
|
||||
handler.NewCol(ActionSequenceCol, e.Sequence()),
|
||||
handler.NewCol(ActionNameCol, e.Name),
|
||||
handler.NewCol(ActionScriptCol, e.Script),
|
||||
handler.NewCol(ActionTimeoutCol, e.Timeout),
|
||||
handler.NewCol(ActionAllowedToFailCol, e.AllowedToFail),
|
||||
handler.NewCol(ActionStateCol, domain.ActionStateActive),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
@@ -93,30 +95,30 @@ func (p *ActionProjection) reduceActionAdded(event eventstore.EventReader) (*han
|
||||
func (p *ActionProjection) reduceActionChanged(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*action.ChangedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-q4oq8", "seq", event.Sequence, "expected", action.ChangedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-Bg8oM", "reduce.wrong.event.type")
|
||||
logging.LogWithFields("HANDL-Dg2th", "seq", event.Sequence, "expected", action.ChangedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-Gg43d", "reduce.wrong.event.type")
|
||||
}
|
||||
values := []handler.Column{
|
||||
handler.NewCol(actionChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(actionSequenceCol, e.Sequence()),
|
||||
handler.NewCol(ActionChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(ActionSequenceCol, e.Sequence()),
|
||||
}
|
||||
if e.Name != nil {
|
||||
values = append(values, handler.NewCol(actionNameCol, *e.Name))
|
||||
values = append(values, handler.NewCol(ActionNameCol, *e.Name))
|
||||
}
|
||||
if e.Script != nil {
|
||||
values = append(values, handler.NewCol(actionScriptCol, *e.Script))
|
||||
values = append(values, handler.NewCol(ActionScriptCol, *e.Script))
|
||||
}
|
||||
if e.Timeout != nil {
|
||||
values = append(values, handler.NewCol(actionTimeoutCol, *e.Timeout))
|
||||
values = append(values, handler.NewCol(ActionTimeoutCol, *e.Timeout))
|
||||
}
|
||||
if e.AllowedToFail != nil {
|
||||
values = append(values, handler.NewCol(actionAllowedToFailCol, *e.AllowedToFail))
|
||||
values = append(values, handler.NewCol(ActionAllowedToFailCol, *e.AllowedToFail))
|
||||
}
|
||||
return crdb.NewUpdateStatement(
|
||||
e,
|
||||
values,
|
||||
[]handler.Condition{
|
||||
handler.NewCond(actionIDCol, e.Aggregate().ID),
|
||||
handler.NewCond(ActionIDCol, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
@@ -124,18 +126,18 @@ func (p *ActionProjection) reduceActionChanged(event eventstore.EventReader) (*h
|
||||
func (p *ActionProjection) reduceActionDeactivated(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*action.DeactivatedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-1gwdc", "seq", event.Sequence, "expectedType", action.DeactivatedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-BApK4", "reduce.wrong.event.type")
|
||||
logging.LogWithFields("HANDL-Fhhjd", "seq", event.Sequence, "expectedType", action.DeactivatedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-Fgh32", "reduce.wrong.event.type")
|
||||
}
|
||||
return crdb.NewUpdateStatement(
|
||||
e,
|
||||
[]handler.Column{
|
||||
handler.NewCol(actionChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(actionSequenceCol, e.Sequence()),
|
||||
handler.NewCol(actionStateCol, domain.ActionStateInactive),
|
||||
handler.NewCol(ActionChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(ActionSequenceCol, e.Sequence()),
|
||||
handler.NewCol(ActionStateCol, domain.ActionStateInactive),
|
||||
},
|
||||
[]handler.Condition{
|
||||
handler.NewCond(actionIDCol, e.Aggregate().ID),
|
||||
handler.NewCond(ActionIDCol, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
@@ -143,18 +145,18 @@ func (p *ActionProjection) reduceActionDeactivated(event eventstore.EventReader)
|
||||
func (p *ActionProjection) reduceActionReactivated(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*action.ReactivatedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-Vjwiy", "seq", event.Sequence, "expectedType", action.ReactivatedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-o37De", "reduce.wrong.event.type")
|
||||
logging.LogWithFields("HANDL-Fg4r3", "seq", event.Sequence, "expectedType", action.ReactivatedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-hwdqa", "reduce.wrong.event.type")
|
||||
}
|
||||
return crdb.NewUpdateStatement(
|
||||
e,
|
||||
[]handler.Column{
|
||||
handler.NewCol(actionChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(actionSequenceCol, e.Sequence()),
|
||||
handler.NewCol(actionStateCol, domain.ActionStateActive),
|
||||
handler.NewCol(ActionChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(ActionSequenceCol, e.Sequence()),
|
||||
handler.NewCol(ActionStateCol, domain.ActionStateActive),
|
||||
},
|
||||
[]handler.Condition{
|
||||
handler.NewCond(actionIDCol, e.Aggregate().ID),
|
||||
handler.NewCond(ActionIDCol, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
@@ -162,13 +164,13 @@ func (p *ActionProjection) reduceActionReactivated(event eventstore.EventReader)
|
||||
func (p *ActionProjection) reduceActionRemoved(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*action.RemovedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-79OhB", "seq", event.Sequence, "expectedType", action.RemovedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-4TbKT", "reduce.wrong.event.type")
|
||||
logging.LogWithFields("HANDL-Dgwh2", "seq", event.Sequence, "expectedType", action.RemovedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-Dgh2d", "reduce.wrong.event.type")
|
||||
}
|
||||
return crdb.NewDeleteStatement(
|
||||
e,
|
||||
[]handler.Condition{
|
||||
handler.NewCond(actionIDCol, e.Aggregate().ID),
|
||||
handler.NewCond(ActionIDCol, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
|
||||
178
internal/query/projection/action_test.go
Normal file
178
internal/query/projection/action_test.go
Normal file
@@ -0,0 +1,178 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/caos/zitadel/internal/domain"
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore"
|
||||
"github.com/caos/zitadel/internal/eventstore/handler"
|
||||
"github.com/caos/zitadel/internal/eventstore/repository"
|
||||
"github.com/caos/zitadel/internal/repository/action"
|
||||
)
|
||||
|
||||
func TestActionProjection_reduces(t *testing.T) {
|
||||
type args struct {
|
||||
event func(t *testing.T) eventstore.EventReader
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
reduce func(event eventstore.EventReader) (*handler.Statement, error)
|
||||
want wantReduce
|
||||
}{
|
||||
{
|
||||
name: "reduceActionAdded",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(action.AddedEventType),
|
||||
action.AggregateType,
|
||||
[]byte(`{"name": "name", "script":"name(){}","timeout": 3000000000, "allowedToFail": true}`),
|
||||
), action.AddedEventMapper),
|
||||
},
|
||||
reduce: (&ActionProjection{}).reduceActionAdded,
|
||||
want: wantReduce{
|
||||
projection: ActionTable,
|
||||
aggregateType: eventstore.AggregateType("action"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "INSERT INTO zitadel.projections.actions (id, creation_date, change_date, resource_owner, sequence, name, script, timeout, allowed_to_fail, action_state) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
|
||||
expectedArgs: []interface{}{
|
||||
"agg-id",
|
||||
anyArg{},
|
||||
anyArg{},
|
||||
"ro-id",
|
||||
uint64(15),
|
||||
"name",
|
||||
"name(){}",
|
||||
3 * time.Second,
|
||||
true,
|
||||
domain.ActionStateActive,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceActionChanged",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(action.ChangedEventType),
|
||||
action.AggregateType,
|
||||
[]byte(`{"name": "name2", "script":"name2(){}"}`),
|
||||
), action.ChangedEventMapper),
|
||||
},
|
||||
reduce: (&ActionProjection{}).reduceActionChanged,
|
||||
want: wantReduce{
|
||||
projection: ActionTable,
|
||||
aggregateType: eventstore.AggregateType("action"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "UPDATE zitadel.projections.actions SET (change_date, sequence, name, script) = ($1, $2, $3, $4) WHERE (id = $5)",
|
||||
expectedArgs: []interface{}{
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
"name2",
|
||||
"name2(){}",
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceActionDeactivated",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(action.ChangedEventType),
|
||||
action.AggregateType,
|
||||
[]byte(`{}`),
|
||||
), action.DeactivatedEventMapper),
|
||||
},
|
||||
reduce: (&ActionProjection{}).reduceActionDeactivated,
|
||||
want: wantReduce{
|
||||
projection: ActionTable,
|
||||
aggregateType: eventstore.AggregateType("action"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "UPDATE zitadel.projections.actions SET (change_date, sequence, action_state) = ($1, $2, $3) WHERE (id = $4)",
|
||||
expectedArgs: []interface{}{
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
domain.ActionStateInactive,
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceActionReactivated",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(action.ChangedEventType),
|
||||
action.AggregateType,
|
||||
[]byte(`{}`),
|
||||
), action.ReactivatedEventMapper),
|
||||
},
|
||||
reduce: (&ActionProjection{}).reduceActionReactivated,
|
||||
want: wantReduce{
|
||||
projection: ActionTable,
|
||||
aggregateType: eventstore.AggregateType("action"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "UPDATE zitadel.projections.actions SET (change_date, sequence, action_state) = ($1, $2, $3) WHERE (id = $4)",
|
||||
expectedArgs: []interface{}{
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
domain.ActionStateActive,
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceActionRemoved",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(action.ChangedEventType),
|
||||
action.AggregateType,
|
||||
[]byte(`{}`),
|
||||
), action.RemovedEventMapper),
|
||||
},
|
||||
reduce: (&ActionProjection{}).reduceActionRemoved,
|
||||
want: wantReduce{
|
||||
projection: ActionTable,
|
||||
aggregateType: eventstore.AggregateType("action"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "DELETE FROM zitadel.projections.actions WHERE (id = $1)",
|
||||
expectedArgs: []interface{}{
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
event := baseEvent(t)
|
||||
got, err := tt.reduce(event)
|
||||
if _, ok := err.(errors.InvalidArgument); !ok {
|
||||
t.Errorf("no wrong event mapping: %v, got: %v", err, got)
|
||||
}
|
||||
|
||||
event = tt.args.event(t)
|
||||
got, err = tt.reduce(event)
|
||||
assertReduce(t, got, err, tt.want)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ type Config struct {
|
||||
BulkLimit uint64
|
||||
CRDB types.SQL
|
||||
Customizations map[string]CustomConfig
|
||||
MaxIterators int
|
||||
}
|
||||
|
||||
type CustomConfig struct {
|
||||
|
||||
93
internal/query/projection/flow.go
Normal file
93
internal/query/projection/flow.go
Normal file
@@ -0,0 +1,93 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/caos/logging"
|
||||
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore"
|
||||
"github.com/caos/zitadel/internal/eventstore/handler"
|
||||
"github.com/caos/zitadel/internal/eventstore/handler/crdb"
|
||||
"github.com/caos/zitadel/internal/repository/org"
|
||||
)
|
||||
|
||||
const (
|
||||
FlowTriggerTable = "zitadel.projections.flows_triggers"
|
||||
FlowTypeCol = "flow_type"
|
||||
FlowTriggerTypeCol = "trigger_type"
|
||||
FlowResourceOwnerCol = "resource_owner"
|
||||
FlowActionTriggerSequenceCol = "trigger_sequence"
|
||||
FlowActionIDCol = "action_id"
|
||||
)
|
||||
|
||||
type FlowProjection struct {
|
||||
crdb.StatementHandler
|
||||
}
|
||||
|
||||
func NewFlowProjection(ctx context.Context, config crdb.StatementHandlerConfig) *FlowProjection {
|
||||
p := &FlowProjection{}
|
||||
config.ProjectionName = FlowTriggerTable
|
||||
config.Reducers = p.reducers()
|
||||
p.StatementHandler = crdb.NewStatementHandler(ctx, config)
|
||||
return p
|
||||
}
|
||||
|
||||
func (p *FlowProjection) reducers() []handler.AggregateReducer {
|
||||
return []handler.AggregateReducer{
|
||||
{
|
||||
Aggregate: org.AggregateType,
|
||||
EventRedusers: []handler.EventReducer{
|
||||
{
|
||||
Event: org.TriggerActionsSetEventType,
|
||||
Reduce: p.reduceTriggerActionsSetEventType,
|
||||
},
|
||||
{
|
||||
Event: org.FlowClearedEventType,
|
||||
Reduce: p.reduceFlowClearedEventType,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (p *FlowProjection) reduceTriggerActionsSetEventType(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*org.TriggerActionsSetEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-zWCk3", "seq", event.Sequence, "expectedType", org.TriggerActionsSetEventType).Error("was not an trigger actions set event")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-uYq4r", "reduce.wrong.event.type")
|
||||
}
|
||||
stmts := make([]func(reader eventstore.EventReader) crdb.Exec, len(e.ActionIDs)+1)
|
||||
stmts[0] = crdb.AddDeleteStatement(
|
||||
[]handler.Condition{
|
||||
handler.NewCond(FlowTypeCol, e.FlowType),
|
||||
handler.NewCond(FlowTriggerTypeCol, e.TriggerType),
|
||||
},
|
||||
)
|
||||
for i, id := range e.ActionIDs {
|
||||
stmts[i+1] = crdb.AddCreateStatement(
|
||||
[]handler.Column{
|
||||
handler.NewCol(FlowResourceOwnerCol, e.Aggregate().ResourceOwner),
|
||||
handler.NewCol(FlowTypeCol, e.FlowType),
|
||||
handler.NewCol(FlowTriggerTypeCol, e.TriggerType),
|
||||
handler.NewCol(FlowActionIDCol, id),
|
||||
handler.NewCol(FlowActionTriggerSequenceCol, i),
|
||||
},
|
||||
)
|
||||
}
|
||||
return crdb.NewMultiStatement(e, stmts...), nil
|
||||
}
|
||||
|
||||
func (p *FlowProjection) reduceFlowClearedEventType(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*org.FlowClearedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-zWCk3", "seq", event.Sequence, "expectedType", org.FlowClearedEventType).Error("was not a flow cleared event")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-uYq4r", "reduce.wrong.event.type")
|
||||
}
|
||||
return crdb.NewDeleteStatement(
|
||||
e,
|
||||
[]handler.Condition{
|
||||
handler.NewCond(FlowTypeCol, e.FlowType),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
93
internal/query/projection/flow_test.go
Normal file
93
internal/query/projection/flow_test.go
Normal file
@@ -0,0 +1,93 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/caos/zitadel/internal/domain"
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore"
|
||||
"github.com/caos/zitadel/internal/eventstore/handler"
|
||||
"github.com/caos/zitadel/internal/eventstore/repository"
|
||||
"github.com/caos/zitadel/internal/repository/org"
|
||||
)
|
||||
|
||||
func TestFlowProjection_reduces(t *testing.T) {
|
||||
type args struct {
|
||||
event func(t *testing.T) eventstore.EventReader
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
reduce func(event eventstore.EventReader) (*handler.Statement, error)
|
||||
want wantReduce
|
||||
}{
|
||||
//TODO: multi stmt tests
|
||||
//{
|
||||
// name: "reduceTriggerActionsSetEventType",
|
||||
// args: args{
|
||||
// event: getEvent(testEvent(
|
||||
// repository.EventType(org.TriggerActionsSetEventType),
|
||||
// org.AggregateType,
|
||||
// []byte(`{"flowType": 1, "triggerType": 1, "actionIDs": ["id1", "id2"]}`),
|
||||
// ), org.TriggerActionsSetEventMapper),
|
||||
// },
|
||||
// reduce: (&FlowProjection{}).reduceTriggerActionsSetEventType,
|
||||
// want: wantReduce{
|
||||
// projection: FlowTriggerTable,
|
||||
// aggregateType: eventstore.AggregateType("org"),
|
||||
// sequence: 15,
|
||||
// previousSequence: 10,
|
||||
// executer: &testExecuter{
|
||||
// shouldExec: true,
|
||||
// expectedStmt: "DELETE FROM zitadel.projections.actions WHERE (flow_type, trigger_type) = ($1, $2); INSERT INTO zitadel.projections.actions (resource_owner, flow_type, trigger_type, action_id, trigger_sequence) = ($3, $1, $2, $4, $5); INSERT INTO zitadel.projections.actions (resource_owner, flow_type, trigger_type, action_id, trigger_sequence) = ($3, $1, $2, $6, $7)",
|
||||
// expectedArgs: []interface{}{
|
||||
// domain.FlowTypeExternalAuthentication,
|
||||
// domain.TriggerTypePostAuthentication,
|
||||
// "ro-id",
|
||||
// "id1",
|
||||
// 0,
|
||||
// "id2",
|
||||
// 1,
|
||||
// },
|
||||
// },
|
||||
// },
|
||||
//},
|
||||
{
|
||||
name: "reduceFlowClearedEventType",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(org.FlowClearedEventType),
|
||||
org.AggregateType,
|
||||
[]byte(`{"flowType": 1}`),
|
||||
), org.FlowClearedEventMapper),
|
||||
},
|
||||
reduce: (&FlowProjection{}).reduceFlowClearedEventType,
|
||||
want: wantReduce{
|
||||
projection: FlowTriggerTable,
|
||||
aggregateType: eventstore.AggregateType("org"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "DELETE FROM zitadel.projections.flows_triggers WHERE (flow_type = $1)",
|
||||
expectedArgs: []interface{}{
|
||||
domain.FlowTypeExternalAuthentication,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
event := baseEvent(t)
|
||||
got, err := tt.reduce(event)
|
||||
if _, ok := err.(errors.InvalidArgument); !ok {
|
||||
t.Errorf("no wrong event mapping: %v, got: %v", err, got)
|
||||
}
|
||||
|
||||
event = tt.args.event(t)
|
||||
got, err = tt.reduce(event)
|
||||
assertReduce(t, got, err, tt.want)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
|
||||
"github.com/caos/logging"
|
||||
|
||||
"github.com/caos/zitadel/internal/domain"
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore"
|
||||
@@ -16,9 +17,13 @@ type OrgProjection struct {
|
||||
crdb.StatementHandler
|
||||
}
|
||||
|
||||
const (
|
||||
OrgProjectionTable = "zitadel.projections.orgs"
|
||||
)
|
||||
|
||||
func NewOrgProjection(ctx context.Context, config crdb.StatementHandlerConfig) *OrgProjection {
|
||||
p := &OrgProjection{}
|
||||
config.ProjectionName = "projections.orgs"
|
||||
config.ProjectionName = OrgProjectionTable
|
||||
config.Reducers = p.reducers()
|
||||
p.StatementHandler = crdb.NewStatementHandler(ctx, config)
|
||||
return p
|
||||
@@ -54,33 +59,35 @@ func (p *OrgProjection) reducers() []handler.AggregateReducer {
|
||||
}
|
||||
}
|
||||
|
||||
type OrgColumn string
|
||||
|
||||
const (
|
||||
orgIDCol = "id"
|
||||
orgCreationDateCol = "creation_date"
|
||||
orgChangeDateCol = "change_date"
|
||||
orgResourceOwnerCol = "resource_owner"
|
||||
orgStateCol = "org_state"
|
||||
orgSequenceCol = "sequence"
|
||||
orgDomainCol = "domain"
|
||||
orgNameCol = "name"
|
||||
OrgColumnID = "id"
|
||||
OrgColumnCreationDate = "creation_date"
|
||||
OrgColumnChangeDate = "change_date"
|
||||
OrgColumnResourceOwner = "resource_owner"
|
||||
OrgColumnState = "org_state"
|
||||
OrgColumnSequence = "sequence"
|
||||
OrgColumnName = "name"
|
||||
OrgColumnDomain = "primary_domain"
|
||||
)
|
||||
|
||||
func (p *OrgProjection) reduceOrgAdded(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*org.OrgAddedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-zWCk3", "seq", event.Sequence, "expectedType", org.OrgAddedEventType).Error("was not an event")
|
||||
logging.LogWithFields("HANDL-zWCk3", "seq", event.Sequence(), "expectedType", org.OrgAddedEventType).Error("was not an event")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-uYq4r", "reduce.wrong.event.type")
|
||||
}
|
||||
return crdb.NewCreateStatement(
|
||||
e,
|
||||
[]handler.Column{
|
||||
handler.NewCol(orgIDCol, e.Aggregate().ID),
|
||||
handler.NewCol(orgCreationDateCol, e.CreationDate()),
|
||||
handler.NewCol(orgChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(orgResourceOwnerCol, e.Aggregate().ResourceOwner),
|
||||
handler.NewCol(orgSequenceCol, e.Sequence()),
|
||||
handler.NewCol(orgNameCol, e.Name),
|
||||
handler.NewCol(orgStateCol, domain.OrgStateActive),
|
||||
handler.NewCol(OrgColumnID, e.Aggregate().ID),
|
||||
handler.NewCol(OrgColumnCreationDate, e.CreationDate()),
|
||||
handler.NewCol(OrgColumnChangeDate, e.CreationDate()),
|
||||
handler.NewCol(OrgColumnResourceOwner, e.Aggregate().ResourceOwner),
|
||||
handler.NewCol(OrgColumnSequence, e.Sequence()),
|
||||
handler.NewCol(OrgColumnName, e.Name),
|
||||
handler.NewCol(OrgColumnState, domain.OrgStateActive),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
@@ -88,21 +95,21 @@ func (p *OrgProjection) reduceOrgAdded(event eventstore.EventReader) (*handler.S
|
||||
func (p *OrgProjection) reduceOrgChanged(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*org.OrgChangedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-q4oq8", "seq", event.Sequence, "expected", org.OrgChangedEventType).Error("wrong event type")
|
||||
logging.LogWithFields("HANDL-q4oq8", "seq", event.Sequence(), "expected", org.OrgChangedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-Bg8oM", "reduce.wrong.event.type")
|
||||
}
|
||||
values := []handler.Column{
|
||||
handler.NewCol(orgChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(orgSequenceCol, e.Sequence()),
|
||||
}
|
||||
if e.Name != "" {
|
||||
values = append(values, handler.NewCol(orgNameCol, e.Name))
|
||||
if e.Name == "" {
|
||||
return crdb.NewNoOpStatement(e), nil
|
||||
}
|
||||
return crdb.NewUpdateStatement(
|
||||
e,
|
||||
values,
|
||||
[]handler.Column{
|
||||
handler.NewCol(OrgColumnChangeDate, e.CreationDate()),
|
||||
handler.NewCol(OrgColumnSequence, e.Sequence()),
|
||||
handler.NewCol(OrgColumnName, e.Name),
|
||||
},
|
||||
[]handler.Condition{
|
||||
handler.NewCond(orgIDCol, e.Aggregate().ID),
|
||||
handler.NewCond(OrgColumnID, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
@@ -110,18 +117,18 @@ func (p *OrgProjection) reduceOrgChanged(event eventstore.EventReader) (*handler
|
||||
func (p *OrgProjection) reduceOrgDeactivated(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*org.OrgDeactivatedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-1gwdc", "seq", event.Sequence, "expectedType", org.OrgDeactivatedEventType).Error("wrong event type")
|
||||
logging.LogWithFields("HANDL-1gwdc", "seq", event.Sequence(), "expectedType", org.OrgDeactivatedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-BApK4", "reduce.wrong.event.type")
|
||||
}
|
||||
return crdb.NewUpdateStatement(
|
||||
e,
|
||||
[]handler.Column{
|
||||
handler.NewCol(orgChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(orgSequenceCol, e.Sequence()),
|
||||
handler.NewCol(orgStateCol, domain.OrgStateInactive),
|
||||
handler.NewCol(OrgColumnChangeDate, e.CreationDate()),
|
||||
handler.NewCol(OrgColumnSequence, e.Sequence()),
|
||||
handler.NewCol(OrgColumnState, domain.OrgStateInactive),
|
||||
},
|
||||
[]handler.Condition{
|
||||
handler.NewCond(orgIDCol, e.Aggregate().ID),
|
||||
handler.NewCond(OrgColumnID, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
@@ -129,18 +136,18 @@ func (p *OrgProjection) reduceOrgDeactivated(event eventstore.EventReader) (*han
|
||||
func (p *OrgProjection) reduceOrgReactivated(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*org.OrgReactivatedEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-Vjwiy", "seq", event.Sequence, "expectedType", org.OrgReactivatedEventType).Error("wrong event type")
|
||||
logging.LogWithFields("HANDL-Vjwiy", "seq", event.Sequence(), "expectedType", org.OrgReactivatedEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-o37De", "reduce.wrong.event.type")
|
||||
}
|
||||
return crdb.NewUpdateStatement(
|
||||
e,
|
||||
[]handler.Column{
|
||||
handler.NewCol(orgChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(orgSequenceCol, e.Sequence()),
|
||||
handler.NewCol(orgStateCol, domain.OrgStateActive),
|
||||
handler.NewCol(OrgColumnChangeDate, e.CreationDate()),
|
||||
handler.NewCol(OrgColumnSequence, e.Sequence()),
|
||||
handler.NewCol(OrgColumnState, domain.OrgStateActive),
|
||||
},
|
||||
[]handler.Condition{
|
||||
handler.NewCond(orgIDCol, e.Aggregate().ID),
|
||||
handler.NewCond(OrgColumnID, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
@@ -148,18 +155,18 @@ func (p *OrgProjection) reduceOrgReactivated(event eventstore.EventReader) (*han
|
||||
func (p *OrgProjection) reducePrimaryDomainSet(event eventstore.EventReader) (*handler.Statement, error) {
|
||||
e, ok := event.(*org.DomainPrimarySetEvent)
|
||||
if !ok {
|
||||
logging.LogWithFields("HANDL-79OhB", "seq", event.Sequence, "expectedType", org.OrgDomainPrimarySetEventType).Error("wrong event type")
|
||||
logging.LogWithFields("HANDL-79OhB", "seq", event.Sequence(), "expectedType", org.OrgDomainPrimarySetEventType).Error("wrong event type")
|
||||
return nil, errors.ThrowInvalidArgument(nil, "HANDL-4TbKT", "reduce.wrong.event.type")
|
||||
}
|
||||
return crdb.NewUpdateStatement(
|
||||
e,
|
||||
[]handler.Column{
|
||||
handler.NewCol(orgChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(orgSequenceCol, e.Sequence()),
|
||||
handler.NewCol(orgDomainCol, e.Domain),
|
||||
handler.NewCol(OrgColumnChangeDate, e.CreationDate()),
|
||||
handler.NewCol(OrgColumnSequence, e.Sequence()),
|
||||
handler.NewCol(OrgColumnDomain, e.Domain),
|
||||
},
|
||||
[]handler.Condition{
|
||||
handler.NewCond(orgIDCol, e.Aggregate().ID),
|
||||
handler.NewCond(OrgColumnID, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
|
||||
@@ -16,15 +16,15 @@ import (
|
||||
)
|
||||
|
||||
type OrgOwner struct {
|
||||
OrgID string `col:"org_id"`
|
||||
OrgName string `col:"org_name"`
|
||||
OrgCreationDate time.Time `col:"org_creation_date"`
|
||||
OwnerID string `col:"owner_id"`
|
||||
OwnerLanguage *language.Tag `col:"owner_language"`
|
||||
OwnerEmailAddress string `col:"owner_email"`
|
||||
OwnerFirstName string `col:"owner_first_name"`
|
||||
OwnerLastName string `col:"owner_last_name"`
|
||||
OwnerGender domain.Gender `col:"owner_gender"`
|
||||
OrgID string
|
||||
OrgName string
|
||||
OrgCreationDate time.Time
|
||||
OwnerID string
|
||||
OwnerLanguage *language.Tag
|
||||
OwnerEmailAddress string
|
||||
OwnerFirstName string
|
||||
OwnerLastName string
|
||||
OwnerGender domain.Gender
|
||||
}
|
||||
|
||||
type OrgOwnerProjection struct {
|
||||
@@ -49,7 +49,7 @@ const (
|
||||
|
||||
func NewOrgOwnerProjection(ctx context.Context, config crdb.StatementHandlerConfig) *OrgOwnerProjection {
|
||||
p := &OrgOwnerProjection{}
|
||||
config.ProjectionName = "projections.org_owners"
|
||||
config.ProjectionName = "zitadel.projections.org_owners"
|
||||
config.Reducers = p.reducers()
|
||||
p.StatementHandler = crdb.NewStatementHandler(ctx, config)
|
||||
return p
|
||||
|
||||
196
internal/query/projection/org_test.go
Normal file
196
internal/query/projection/org_test.go
Normal file
@@ -0,0 +1,196 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/caos/zitadel/internal/domain"
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore"
|
||||
"github.com/caos/zitadel/internal/eventstore/handler"
|
||||
"github.com/caos/zitadel/internal/eventstore/repository"
|
||||
"github.com/caos/zitadel/internal/repository/org"
|
||||
)
|
||||
|
||||
func TestOrgProjection_reduces(t *testing.T) {
|
||||
type args struct {
|
||||
event func(t *testing.T) eventstore.EventReader
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
reduce func(event eventstore.EventReader) (*handler.Statement, error)
|
||||
want wantReduce
|
||||
}{
|
||||
{
|
||||
name: "reducePrimaryDomainSet",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(org.OrgDomainPrimarySetEventType),
|
||||
org.AggregateType,
|
||||
[]byte(`{"domain": "domain.new"}`),
|
||||
), org.DomainPrimarySetEventMapper),
|
||||
},
|
||||
reduce: (&OrgProjection{}).reducePrimaryDomainSet,
|
||||
want: wantReduce{
|
||||
projection: OrgProjectionTable,
|
||||
aggregateType: eventstore.AggregateType("org"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "UPDATE zitadel.projections.orgs SET (change_date, sequence, primary_domain) = ($1, $2, $3) WHERE (id = $4)",
|
||||
expectedArgs: []interface{}{
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
"domain.new",
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceOrgReactivated",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(org.OrgReactivatedEventType),
|
||||
org.AggregateType,
|
||||
nil,
|
||||
), org.OrgReactivatedEventMapper),
|
||||
},
|
||||
reduce: (&OrgProjection{}).reduceOrgReactivated,
|
||||
want: wantReduce{
|
||||
projection: OrgProjectionTable,
|
||||
aggregateType: eventstore.AggregateType("org"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "UPDATE zitadel.projections.orgs SET (change_date, sequence, org_state) = ($1, $2, $3) WHERE (id = $4)",
|
||||
expectedArgs: []interface{}{
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
domain.OrgStateActive,
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceOrgDeactivated",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(org.OrgDeactivatedEventType),
|
||||
org.AggregateType,
|
||||
nil,
|
||||
), org.OrgDeactivatedEventMapper),
|
||||
},
|
||||
reduce: (&OrgProjection{}).reduceOrgDeactivated,
|
||||
want: wantReduce{
|
||||
projection: OrgProjectionTable,
|
||||
aggregateType: eventstore.AggregateType("org"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "UPDATE zitadel.projections.orgs SET (change_date, sequence, org_state) = ($1, $2, $3) WHERE (id = $4)",
|
||||
expectedArgs: []interface{}{
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
domain.OrgStateInactive,
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceOrgChanged",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(org.OrgChangedEventType),
|
||||
org.AggregateType,
|
||||
[]byte(`{"name": "new name"}`),
|
||||
), org.OrgChangedEventMapper),
|
||||
},
|
||||
reduce: (&OrgProjection{}).reduceOrgChanged,
|
||||
want: wantReduce{
|
||||
projection: OrgProjectionTable,
|
||||
aggregateType: eventstore.AggregateType("org"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "UPDATE zitadel.projections.orgs SET (change_date, sequence, name) = ($1, $2, $3) WHERE (id = $4)",
|
||||
expectedArgs: []interface{}{
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
"new name",
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceOrgChanged no changes",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(org.OrgChangedEventType),
|
||||
org.AggregateType,
|
||||
[]byte(`{}`),
|
||||
), org.OrgChangedEventMapper),
|
||||
},
|
||||
reduce: (&OrgProjection{}).reduceOrgChanged,
|
||||
want: wantReduce{
|
||||
projection: OrgProjectionTable,
|
||||
aggregateType: eventstore.AggregateType("org"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: false,
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceOrgAdded",
|
||||
args: args{
|
||||
event: getEvent(testEvent(
|
||||
repository.EventType(org.OrgAddedEventType),
|
||||
org.AggregateType,
|
||||
[]byte(`{"name": "name"}`),
|
||||
), org.OrgAddedEventMapper),
|
||||
},
|
||||
reduce: (&OrgProjection{}).reduceOrgAdded,
|
||||
want: wantReduce{
|
||||
projection: OrgProjectionTable,
|
||||
aggregateType: eventstore.AggregateType("org"),
|
||||
sequence: 15,
|
||||
previousSequence: 10,
|
||||
executer: &testExecuter{
|
||||
shouldExec: true,
|
||||
expectedStmt: "INSERT INTO zitadel.projections.orgs (id, creation_date, change_date, resource_owner, sequence, name, org_state) VALUES ($1, $2, $3, $4, $5, $6, $7)",
|
||||
expectedArgs: []interface{}{
|
||||
"agg-id",
|
||||
anyArg{},
|
||||
anyArg{},
|
||||
"ro-id",
|
||||
uint64(15),
|
||||
"name",
|
||||
domain.OrgStateActive,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
event := baseEvent(t)
|
||||
got, err := tt.reduce(event)
|
||||
if _, ok := err.(errors.InvalidArgument); !ok {
|
||||
t.Errorf("no wrong event mapping: %v, got: %v", err, got)
|
||||
}
|
||||
|
||||
event = tt.args.event(t)
|
||||
got, err = tt.reduce(event)
|
||||
assertReduce(t, got, err, tt.want)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -15,7 +15,7 @@ type ProjectProjection struct {
|
||||
|
||||
func NewProjectProjection(ctx context.Context, config crdb.StatementHandlerConfig) *ProjectProjection {
|
||||
p := &ProjectProjection{}
|
||||
config.ProjectionName = "projections.projects"
|
||||
config.ProjectionName = "zitadel.projections.projects"
|
||||
config.Reducers = p.reducers()
|
||||
p.StatementHandler = crdb.NewStatementHandler(ctx, config)
|
||||
return p
|
||||
|
||||
@@ -2,25 +2,21 @@ package projection
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"time"
|
||||
|
||||
"github.com/caos/zitadel/internal/eventstore"
|
||||
"github.com/caos/zitadel/internal/eventstore/handler"
|
||||
"github.com/caos/zitadel/internal/eventstore/handler/crdb"
|
||||
"github.com/caos/zitadel/internal/query/projection/flow"
|
||||
)
|
||||
|
||||
const (
|
||||
currentSeqTable = "projections.current_sequences"
|
||||
CurrentSeqTable = "projections.current_sequences"
|
||||
locksTable = "projections.locks"
|
||||
failedEventsTable = "projections.failed_events"
|
||||
)
|
||||
|
||||
func Start(ctx context.Context, es *eventstore.Eventstore, config Config) error {
|
||||
sqlClient, err := config.CRDB.Start()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
func Start(ctx context.Context, sqlClient *sql.DB, es *eventstore.Eventstore, config Config) error {
|
||||
projectionConfig := crdb.StatementHandlerConfig{
|
||||
ProjectionHandlerConfig: handler.ProjectionHandlerConfig{
|
||||
HandlerConfig: handler.HandlerConfig{
|
||||
@@ -30,19 +26,18 @@ func Start(ctx context.Context, es *eventstore.Eventstore, config Config) error
|
||||
RetryFailedAfter: config.RetryFailedAfter.Duration,
|
||||
},
|
||||
Client: sqlClient,
|
||||
SequenceTable: currentSeqTable,
|
||||
SequenceTable: CurrentSeqTable,
|
||||
LockTable: locksTable,
|
||||
FailedEventsTable: failedEventsTable,
|
||||
MaxFailureCount: config.MaxFailureCount,
|
||||
BulkLimit: config.BulkLimit,
|
||||
}
|
||||
|
||||
// turned off for this release
|
||||
//NewOrgProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["orgs"]))
|
||||
NewOrgProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["orgs"]))
|
||||
//NewProjectProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["projects"]))
|
||||
//owner.NewOrgOwnerProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["org_owners"]))
|
||||
NewActionProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["actions"]))
|
||||
flow.NewFlowProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["flows"]))
|
||||
NewFlowProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["flows"]))
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -62,3 +57,20 @@ func applyCustomConfig(config crdb.StatementHandlerConfig, customConfig CustomCo
|
||||
|
||||
return config
|
||||
}
|
||||
|
||||
func iteratorPool(workerCount int) chan func() {
|
||||
if workerCount <= 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
queue := make(chan func())
|
||||
for i := 0; i < workerCount; i++ {
|
||||
go func() {
|
||||
for iteration := range queue {
|
||||
iteration()
|
||||
time.Sleep(2 * time.Second)
|
||||
}
|
||||
}()
|
||||
}
|
||||
return queue
|
||||
}
|
||||
|
||||
87
internal/query/projection/test_event.go
Normal file
87
internal/query/projection/test_event.go
Normal file
@@ -0,0 +1,87 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/caos/zitadel/internal/eventstore"
|
||||
"github.com/caos/zitadel/internal/eventstore/handler"
|
||||
"github.com/caos/zitadel/internal/eventstore/repository"
|
||||
)
|
||||
|
||||
func testEvent(
|
||||
eventType repository.EventType,
|
||||
aggregateType repository.AggregateType,
|
||||
data []byte,
|
||||
) *repository.Event {
|
||||
return &repository.Event{
|
||||
Sequence: 15,
|
||||
PreviousAggregateSequence: 10,
|
||||
PreviousAggregateTypeSequence: 10,
|
||||
CreationDate: time.Now(),
|
||||
Type: eventType,
|
||||
AggregateType: aggregateType,
|
||||
Data: data,
|
||||
Version: "v1",
|
||||
AggregateID: "agg-id",
|
||||
ResourceOwner: "ro-id",
|
||||
ID: "event-id",
|
||||
EditorService: "editor-svc",
|
||||
EditorUser: "editor-user",
|
||||
}
|
||||
}
|
||||
|
||||
func baseEvent(*testing.T) eventstore.EventReader {
|
||||
return &eventstore.BaseEvent{}
|
||||
}
|
||||
|
||||
func getEvent(event *repository.Event, mapper func(*repository.Event) (eventstore.EventReader, error)) func(t *testing.T) eventstore.EventReader {
|
||||
return func(t *testing.T) eventstore.EventReader {
|
||||
e, err := mapper(event)
|
||||
if err != nil {
|
||||
t.Fatalf("mapper failed: %v", err)
|
||||
}
|
||||
return e
|
||||
}
|
||||
}
|
||||
|
||||
type wantReduce struct {
|
||||
projection string
|
||||
aggregateType eventstore.AggregateType
|
||||
sequence uint64
|
||||
previousSequence uint64
|
||||
executer *testExecuter
|
||||
err func(error) bool
|
||||
}
|
||||
|
||||
func assertReduce(t *testing.T, stmt *handler.Statement, err error, want wantReduce) {
|
||||
t.Helper()
|
||||
if want.err == nil && err != nil {
|
||||
t.Errorf("unexpected error of type %T: %v", err, err)
|
||||
return
|
||||
}
|
||||
if want.err != nil && want.err(err) {
|
||||
return
|
||||
}
|
||||
if stmt.AggregateType != want.aggregateType {
|
||||
t.Errorf("wront aggregate type: want: %q got: %q", want.aggregateType, stmt.AggregateType)
|
||||
}
|
||||
|
||||
if stmt.PreviousSequence != want.previousSequence {
|
||||
t.Errorf("wront previous sequence: want: %d got: %d", want.previousSequence, stmt.PreviousSequence)
|
||||
}
|
||||
|
||||
if stmt.Sequence != want.sequence {
|
||||
t.Errorf("wront sequence: want: %d got: %d", want.sequence, stmt.Sequence)
|
||||
}
|
||||
if stmt.Execute == nil {
|
||||
want.executer.Validate(t)
|
||||
return
|
||||
}
|
||||
err = stmt.Execute(want.executer, want.projection)
|
||||
if err != nil {
|
||||
t.Errorf("unexpected error: %v", err)
|
||||
}
|
||||
|
||||
want.executer.Validate(t)
|
||||
}
|
||||
48
internal/query/projection/test_executer.go
Normal file
48
internal/query/projection/test_executer.go
Normal file
@@ -0,0 +1,48 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"testing"
|
||||
)
|
||||
|
||||
type testExecuter struct {
|
||||
expectedStmt string
|
||||
gottenStmt string
|
||||
shouldExec bool
|
||||
|
||||
expectedArgs []interface{}
|
||||
gottenArgs []interface{}
|
||||
gotExecuted bool
|
||||
}
|
||||
|
||||
type anyArg struct{}
|
||||
|
||||
func (e *testExecuter) Exec(stmt string, args ...interface{}) (sql.Result, error) {
|
||||
e.gottenStmt = stmt
|
||||
e.gottenArgs = args
|
||||
e.gotExecuted = true
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (e *testExecuter) Validate(t *testing.T) {
|
||||
t.Helper()
|
||||
if e.shouldExec != e.gotExecuted {
|
||||
t.Error("expected to be executed")
|
||||
return
|
||||
}
|
||||
if len(e.gottenArgs) != len(e.expectedArgs) {
|
||||
t.Errorf("wrong arg len expected: %d got: %d", len(e.expectedArgs), len(e.gottenArgs))
|
||||
} else {
|
||||
for i := 0; i < len(e.expectedArgs); i++ {
|
||||
if _, ok := e.expectedArgs[i].(anyArg); ok {
|
||||
continue
|
||||
}
|
||||
if e.expectedArgs[i] != e.gottenArgs[i] {
|
||||
t.Errorf("wrong argument at index %d: got: %v want: %v", i, e.gottenArgs[i], e.expectedArgs[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
if e.gottenStmt != e.expectedStmt {
|
||||
t.Errorf("wrong stmt want:\n%s\ngot:\n%s", e.expectedStmt, e.gottenStmt)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user