feat: org queries (#136)

* search orgs

* org by domain

* member spooler

* member

* get roles

* tests

* types duration

* use default func for renew

* correct database

* reorder migrations

* delete unused consts

* move get roles to internal

* use prepared org by domain

* implement org in other objects

* add eventstores

internal/admin/repository/eventsourcing/eventstore/org.go

@@ -4,11 +4,12 @@ import (
 	"context"
 
 	admin_model "github.com/caos/zitadel/internal/admin/model"
-	"github.com/caos/zitadel/internal/errors"
+	admin_view "github.com/caos/zitadel/internal/admin/repository/eventsourcing/view"
 	"github.com/caos/zitadel/internal/eventstore"
 	"github.com/caos/zitadel/internal/eventstore/sdk"
 	org_model "github.com/caos/zitadel/internal/org/model"
 	org_es "github.com/caos/zitadel/internal/org/repository/eventsourcing"
+	org_view "github.com/caos/zitadel/internal/org/repository/view"
 	usr_es "github.com/caos/zitadel/internal/user/repository/eventsourcing"
 )
 
@@ -16,6 +17,10 @@ type OrgRepo struct {
 	Eventstore     eventstore.Eventstore
 	OrgEventstore  *org_es.OrgEventstore
 	UserEventstore *usr_es.UserEventstore
+
+	View *admin_view.View
+
+	SearchLimit uint64
 }
 
 func (repo *OrgRepo) SetUpOrg(ctx context.Context, setUp *admin_model.SetupOrg) (*admin_model.SetupOrg, error) {
@@ -51,8 +56,18 @@ func (repo *OrgRepo) OrgByID(ctx context.Context, id string) (*org_model.Org, er
 	return repo.OrgEventstore.OrgByID(ctx, org_model.NewOrg(id))
 }
 
-func (repo *OrgRepo) SearchOrgs(ctx context.Context) ([]*org_model.Org, error) {
-	return nil, errors.ThrowUnimplemented(nil, "EVENT-hFIHK", "search not implemented")
+func (repo *OrgRepo) SearchOrgs(ctx context.Context, query *org_model.OrgSearchRequest) (*org_model.OrgSearchResult, error) {
+	query.EnsureLimit(repo.SearchLimit)
+	orgs, count, err := repo.View.SearchOrgs(query)
+	if err != nil {
+		return nil, err
+	}
+	return &org_model.OrgSearchResult{
+		Offset:      query.Offset,
+		Limit:       query.Limit,
+		TotalResult: uint64(count),
+		Result:      org_view.OrgsToModel(orgs),
+	}, nil
 }
 
 func (repo *OrgRepo) IsOrgUnique(ctx context.Context, name, domain string) (isUnique bool, err error) {

internal/admin/repository/eventsourcing/handler/handler.go

@@ -0,0 +1,43 @@
+package handler
+
+import (
+	"time"
+
+	"github.com/caos/zitadel/internal/admin/repository/eventsourcing/view"
+	"github.com/caos/zitadel/internal/config/types"
+	"github.com/caos/zitadel/internal/eventstore/spooler"
+	proj_event "github.com/caos/zitadel/internal/project/repository/eventsourcing"
+	usr_event "github.com/caos/zitadel/internal/user/repository/eventsourcing"
+)
+
+type Configs map[string]*Config
+
+type Config struct {
+	MinimumCycleDuration types.Duration
+}
+
+type handler struct {
+	view                *view.View
+	bulkLimit           uint64
+	cycleDuration       time.Duration
+	errorCountUntilSkip uint64
+}
+
+type EventstoreRepos struct {
+	ProjectEvents *proj_event.ProjectEventstore
+	UserEvents    *usr_event.UserEventstore
+}
+
+func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View) []spooler.Handler {
+	return []spooler.Handler{
+		&Org{handler: handler{view, bulkLimit, configs.cycleDuration("Org"), errorCount}},
+	}
+}
+
+func (configs Configs) cycleDuration(viewModel string) time.Duration {
+	c, ok := configs[viewModel]
+	if !ok {
+		return 1 * time.Second
+	}
+	return c.MinimumCycleDuration.Duration
+}

internal/admin/repository/eventsourcing/handler/org.go

@@ -0,0 +1,65 @@
+package handler
+
+import (
+	"time"
+
+	"github.com/caos/logging"
+	es_models "github.com/caos/zitadel/internal/eventstore/models"
+	"github.com/caos/zitadel/internal/eventstore/spooler"
+	org_model "github.com/caos/zitadel/internal/org/model"
+	"github.com/caos/zitadel/internal/org/repository/eventsourcing"
+	"github.com/caos/zitadel/internal/org/repository/view"
+)
+
+type Org struct {
+	handler
+}
+
+const (
+	orgTable = "admin_api.orgs"
+)
+
+func (o *Org) MinimumCycleDuration() time.Duration { return o.cycleDuration }
+
+func (o *Org) ViewModel() string {
+	return orgTable
+}
+
+func (o *Org) EventQuery() (*es_models.SearchQuery, error) {
+	sequence, err := o.view.GetLatestOrgSequence()
+	if err != nil {
+		return nil, err
+	}
+	return eventsourcing.OrgQuery(sequence), nil
+}
+
+func (o *Org) Process(event *es_models.Event) error {
+	org := new(view.OrgView)
+
+	switch event.Type {
+	case org_model.OrgAdded:
+		org.AppendEvent(event)
+	case org_model.OrgChanged:
+		err := org.SetData(event)
+		if err != nil {
+			return err
+		}
+		org, err = o.view.OrgByID(org.ID)
+		if err != nil {
+			return err
+		}
+		err = org.AppendEvent(event)
+		if err != nil {
+			return err
+		}
+	default:
+		return o.view.ProcessedOrgSequence(event.Sequence)
+	}
+
+	return o.view.PutOrg(org)
+}
+
+func (o *Org) OnError(event *es_models.Event, spoolerErr error) error {
+	logging.LogWithFields("SPOOL-ls9ew", "id", event.AggregateID).WithError(spoolerErr).Warn("something went wrong in project app handler")
+	return spooler.HandleError(event, spoolerErr, o.view.GetLatestOrgFailedEvent, o.view.ProcessedOrgFailedEvent, o.view.ProcessedOrgSequence, o.errorCountUntilSkip)
+}

internal/admin/repository/eventsourcing/repository.go

@@ -6,8 +6,12 @@ import (
 	"github.com/caos/logging"
 	"github.com/caos/zitadel/internal/admin/repository/eventsourcing/eventstore"
 	"github.com/caos/zitadel/internal/admin/repository/eventsourcing/setup"
+	"github.com/caos/zitadel/internal/admin/repository/eventsourcing/spooler"
+	admin_view "github.com/caos/zitadel/internal/admin/repository/eventsourcing/view"
 	sd "github.com/caos/zitadel/internal/config/systemdefaults"
+	"github.com/caos/zitadel/internal/config/types"
 	es_int "github.com/caos/zitadel/internal/eventstore"
+	es_spol "github.com/caos/zitadel/internal/eventstore/spooler"
 	es_iam "github.com/caos/zitadel/internal/iam/repository/eventsourcing"
 	es_org "github.com/caos/zitadel/internal/org/repository/eventsourcing"
 	es_proj "github.com/caos/zitadel/internal/project/repository/eventsourcing"
@@ -15,13 +19,14 @@ import (
 )
 
 type Config struct {
-	Eventstore es_int.Config
-	//View       view.ViewConfig
-	//Spooler    spooler.SpoolerConfig
+	SearchLimit uint64
+	Eventstore  es_int.Config
+	View        types.SQL
+	Spooler     spooler.SpoolerConfig
 }
 
 type EsRepository struct {
-	//spooler *es_spooler.Spooler
+	spooler *es_spol.Spooler
 	eventstore.OrgRepo
 }
 
@@ -31,15 +36,6 @@ func Start(ctx context.Context, conf Config, systemDefaults sd.SystemDefaults) (
 		return nil, err
 	}
 
-	//view, sql, err := mgmt_view.StartView(conf.View)
-	//if err != nil {
-	//	return nil, err
-	//}
-
-	//conf.Spooler.View = view
-	//conf.Spooler.EsClient = es.Client
-	//conf.Spooler.SQL = sql
-	//spool := spooler.StartSpooler(conf.Spooler)
 	iam, err := es_iam.StartIam(es_iam.IamConfig{
 		Eventstore: es,
 		Cache:      conf.Eventstore.Cache,
@@ -66,15 +62,29 @@ func Start(ctx context.Context, conf Config, systemDefaults sd.SystemDefaults) (
 		return nil, err
 	}
 
+	sqlClient, err := conf.View.Start()
+	if err != nil {
+		return nil, err
+	}
+	view, err := admin_view.StartView(sqlClient)
+	if err != nil {
+		return nil, err
+	}
+
 	eventstoreRepos := setup.EventstoreRepos{OrgEvents: org, UserEvents: user, ProjectEvents: project, IamEvents: iam}
 	err = setup.StartSetup(systemDefaults, eventstoreRepos).Execute(ctx)
 	logging.Log("SERVE-k280HZ").OnError(err).Panic("failed to execute setup")
 
+	spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient)
+
 	return &EsRepository{
+		spooler: spool,
 		OrgRepo: eventstore.OrgRepo{
 			Eventstore:     es,
 			OrgEventstore:  org,
 			UserEventstore: user,
+			View:           view,
+			SearchLimit:    conf.SearchLimit,
 		},
 	}, nil
 }

internal/admin/repository/eventsourcing/spooler/lock.go

@@ -0,0 +1,29 @@
+package spooler
+
+import (
+	"database/sql"
+	"time"
+
+	es_locker "github.com/caos/zitadel/internal/eventstore/locker"
+)
+
+const (
+	lockTable      = "admin_api.locks"
+	lockedUntilKey = "locked_until"
+	lockerIDKey    = "locker_id"
+	objectTypeKey  = "object_type"
+)
+
+type locker struct {
+	dbClient *sql.DB
+}
+
+type lock struct {
+	LockerID    string    `gorm:"column:locker_id;primary_key"`
+	LockedUntil time.Time `gorm:"column:locked_until"`
+	ViewName    string    `gorm:"column:object_type;primary_key"`
+}
+
+func (l *locker) Renew(lockerID, viewModel string, waitTime time.Duration) error {
+	return es_locker.Renew(l.dbClient, lockTable, lockerID, viewModel, waitTime)
+}

internal/admin/repository/eventsourcing/spooler/lock_test.go

@@ -0,0 +1,127 @@
+package spooler
+
+import (
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+)
+
+type dbMock struct {
+	db   *sql.DB
+	mock sqlmock.Sqlmock
+}
+
+func mockDB(t *testing.T) *dbMock {
+	mockDB := dbMock{}
+	var err error
+	mockDB.db, mockDB.mock, err = sqlmock.New()
+	if err != nil {
+		t.Fatalf("error occured while creating stub db %v", err)
+	}
+
+	mockDB.mock.MatchExpectationsInOrder(true)
+
+	return &mockDB
+}
+
+func (db *dbMock) expectCommit() *dbMock {
+	db.mock.ExpectCommit()
+
+	return db
+}
+
+func (db *dbMock) expectRollback() *dbMock {
+	db.mock.ExpectRollback()
+
+	return db
+}
+
+func (db *dbMock) expectBegin() *dbMock {
+	db.mock.ExpectBegin()
+
+	return db
+}
+
+func (db *dbMock) expectSavepoint() *dbMock {
+	db.mock.ExpectExec("SAVEPOINT").WillReturnResult(sqlmock.NewResult(1, 1))
+	return db
+}
+
+func (db *dbMock) expectReleaseSavepoint() *dbMock {
+	db.mock.ExpectExec("RELEASE SAVEPOINT").WillReturnResult(sqlmock.NewResult(1, 1))
+
+	return db
+}
+
+func (db *dbMock) expectRenew(lockerID, view string, affectedRows int64) *dbMock {
+	query := db.mock.
+		ExpectExec(`INSERT INTO admin_api\.locks \(object_type, locker_id, locked_until\) VALUES \(\$1, \$2, now\(\)\+\$3\) ON CONFLICT \(object_type\) DO UPDATE SET locked_until = now\(\)\+\$4, locker_id = \$5 WHERE \(locks\.locked_until < now\(\) OR locks\.locker_id = \$6\) AND locks\.object_type = \$7`).
+		WithArgs(view, lockerID, sqlmock.AnyArg(), sqlmock.AnyArg(), lockerID, lockerID, view).
+		WillReturnResult(sqlmock.NewResult(1, 1))
+
+	if affectedRows == 0 {
+		query.WillReturnResult(sqlmock.NewResult(0, 0))
+	} else {
+		query.WillReturnResult(sqlmock.NewResult(1, affectedRows))
+	}
+
+	return db
+}
+
+func Test_locker_Renew(t *testing.T) {
+	type fields struct {
+		db *dbMock
+	}
+	type args struct {
+		lockerID  string
+		viewModel string
+		waitTime  time.Duration
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "renew succeeded",
+			fields: fields{
+				db: mockDB(t).
+					expectBegin().
+					expectSavepoint().
+					expectRenew("locker", "view", 1).
+					expectReleaseSavepoint().
+					expectCommit(),
+			},
+			args:    args{lockerID: "locker", viewModel: "view", waitTime: 1 * time.Second},
+			wantErr: false,
+		},
+		{
+			name: "renew now rows updated",
+			fields: fields{
+				db: mockDB(t).
+					expectBegin().
+					expectSavepoint().
+					expectRenew("locker", "view", 0).
+					expectRollback(),
+			},
+			args:    args{lockerID: "locker", viewModel: "view", waitTime: 1 * time.Second},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			l := &locker{
+				dbClient: tt.fields.db.db,
+			}
+			if err := l.Renew(tt.args.lockerID, tt.args.viewModel, tt.args.waitTime); (err != nil) != tt.wantErr {
+				t.Errorf("locker.Renew() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if err := tt.fields.db.mock.ExpectationsWereMet(); err != nil {
+				t.Errorf("not all database expectations met: %v", err)
+			}
+		})
+	}
+}

internal/admin/repository/eventsourcing/spooler/spooler.go

@@ -0,0 +1,29 @@
+package spooler
+
+import (
+	"database/sql"
+
+	"github.com/caos/zitadel/internal/admin/repository/eventsourcing/handler"
+	"github.com/caos/zitadel/internal/admin/repository/eventsourcing/view"
+	"github.com/caos/zitadel/internal/eventstore"
+	"github.com/caos/zitadel/internal/eventstore/spooler"
+)
+
+type SpoolerConfig struct {
+	BulkLimit             uint64
+	FailureCountUntilSkip uint64
+	ConcurrentTasks       int
+	Handlers              handler.Configs
+}
+
+func StartSpooler(c SpoolerConfig, es eventstore.Eventstore, view *view.View, sql *sql.DB) *spooler.Spooler {
+	spoolerConfig := spooler.Config{
+		Eventstore:      es,
+		Locker:          &locker{dbClient: sql},
+		ConcurrentTasks: c.ConcurrentTasks,
+		ViewHandlers:    handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view),
+	}
+	spool := spoolerConfig.New()
+	spool.Start()
+	return spool
+}

internal/admin/repository/eventsourcing/view/error_event.go

@@ -0,0 +1,17 @@
+package view
+
+import (
+	"github.com/caos/zitadel/internal/view"
+)
+
+const (
+	errTable = "admin_api.failed_event"
+)
+
+func (v *View) saveFailedEvent(failedEvent *view.FailedEvent) error {
+	return view.SaveFailedEvent(v.Db, errTable, failedEvent)
+}
+
+func (v *View) latestFailedEvent(viewName string, sequence uint64) (*view.FailedEvent, error) {
+	return view.LatestFailedEvent(v.Db, errTable, viewName, sequence)
+}

internal/admin/repository/eventsourcing/view/org.go

@@ -0,0 +1,43 @@
+package view
+
+import (
+	org_model "github.com/caos/zitadel/internal/org/model"
+	org_view "github.com/caos/zitadel/internal/org/repository/view"
+	"github.com/caos/zitadel/internal/view"
+)
+
+const (
+	orgTable = "admin_api.orgs"
+)
+
+func (v *View) OrgByID(orgID string) (*org_view.OrgView, error) {
+	return org_view.OrgByID(v.Db, orgTable, orgID)
+}
+
+func (v *View) SearchOrgs(query *org_model.OrgSearchRequest) ([]*org_view.OrgView, int, error) {
+	return org_view.SearchOrgs(v.Db, orgTable, query)
+}
+
+func (v *View) PutOrg(org *org_view.OrgView) error {
+	err := org_view.PutOrg(v.Db, orgTable, org)
+	if err != nil {
+		return err
+	}
+	return v.ProcessedOrgSequence(org.Sequence)
+}
+
+func (v *View) GetLatestOrgFailedEvent(sequence uint64) (*view.FailedEvent, error) {
+	return v.latestFailedEvent(orgTable, sequence)
+}
+
+func (v *View) ProcessedOrgFailedEvent(failedEvent *view.FailedEvent) error {
+	return v.saveFailedEvent(failedEvent)
+}
+
+func (v *View) GetLatestOrgSequence() (uint64, error) {
+	return v.latestSequence(orgTable)
+}
+
+func (v *View) ProcessedOrgSequence(eventSequence uint64) error {
+	return v.saveCurrentSequence(orgTable, eventSequence)
+}

internal/admin/repository/eventsourcing/view/sequence.go

@@ -0,0 +1,17 @@
+package view
+
+import (
+	"github.com/caos/zitadel/internal/view"
+)
+
+const (
+	sequencesTable = "admin_api.current_sequences"
+)
+
+func (v *View) saveCurrentSequence(viewName string, sequence uint64) error {
+	return view.SaveCurrentSequence(v.Db, sequencesTable, viewName, sequence)
+}
+
+func (v *View) latestSequence(viewName string) (uint64, error) {
+	return view.LatestSequence(v.Db, sequencesTable, viewName)
+}

internal/admin/repository/eventsourcing/view/view.go

@@ -0,0 +1,25 @@
+package view
+
+import (
+	"database/sql"
+
+	"github.com/jinzhu/gorm"
+)
+
+type View struct {
+	Db *gorm.DB
+}
+
+func StartView(sqlClient *sql.DB) (*View, error) {
+	gorm, err := gorm.Open("postgres", sqlClient)
+	if err != nil {
+		return nil, err
+	}
+	return &View{
+		Db: gorm,
+	}, nil
+}
+
+func (v *View) Health() (err error) {
+	return v.Db.DB().Ping()
+}