refactor: relational organization projection to use repository (#11621)

- Transitioned from using `handler.Statements` to a repository pattern
for managing organization data in migrations.
- Introduced a new method to set the updated timestamp for
organizations.
- Updated the projection logic to utilize the repository for creating,
updating, and deleting organization records.

backend/v3/domain/organization.go

@@ -71,8 +71,10 @@ type organizationConditions interface {
 type organizationChanges interface {
 	// SetName sets the name column.
 	SetName(name string) database.Change
-	// SetState sets the name column.
+	// SetState sets the state column.
 	SetState(state OrgState) database.Change
+	// SetUpdatedAt sets the updated at column.
+	SetUpdatedAt(updatedAt time.Time) database.Change
 }
 
 //go:generate mockgen -typed -package domainmock -destination ./mock/org.mock.go . OrganizationRepository

backend/v3/storage/database/repository/idp_intent.go

@@ -69,7 +69,9 @@ func (i idpIntentRepository) Create(ctx context.Context, client database.QueryEx
 	}
 
 	builder := new(database.StatementBuilder)
-	builder.WriteString(`INSERT INTO ` + i.qualifiedTableName() + ` (instance_id, id, success_url, failure_url, idp_id, idp_arguments, created_at, updated_at) VALUES ( `)
+	builder.WriteString(`INSERT INTO `)
+	builder.WriteString(i.qualifiedTableName())
+	builder.WriteString(` (instance_id, id, success_url, failure_url, idp_id, idp_arguments, created_at, updated_at) VALUES ( `)
 	builder.WriteArgs(intent.InstanceID, intent.ID, intent.SuccessURL.String(), intent.FailureURL.String(), intent.IDPID, intent.IDPArguments, createdAt, updatedAt)
 	builder.WriteString(` ) RETURNING created_at, updated_at`)
 	return client.QueryRow(ctx, builder.String(), builder.Args()...).Scan(&intent.CreatedAt, &intent.UpdatedAt)

backend/v3/storage/database/repository/org.go

@@ -2,6 +2,7 @@ package repository
 
 import (
 	"context"
+	"time"
 
 	"github.com/zitadel/zitadel/backend/v3/domain"
 	"github.com/zitadel/zitadel/backend/v3/storage/database"
@@ -24,6 +25,10 @@ func (o org) unqualifiedTableName() string {
 	return "organizations"
 }
 
+func (o org) qualifiedTableName() string {
+	return "zitadel.organizations"
+}
+
 func OrganizationRepository() domain.OrganizationRepository {
 	return new(org)
 }
@@ -81,16 +86,25 @@ func (o org) List(ctx context.Context, client database.QueryExecutor, opts ...da
 	return scanOrganizations(ctx, client, &builder)
 }
 
-const createOrganizationStmt = `INSERT INTO zitadel.organizations (id, name, instance_id, state)` +
-	` VALUES ($1, $2, $3, $4)` +
-	` RETURNING created_at, updated_at`
-
 // Create implements [domain.OrganizationRepository].
 func (o org) Create(ctx context.Context, client database.QueryExecutor, organization *domain.Organization) error {
-	builder := database.StatementBuilder{}
-	builder.AppendArgs(organization.ID, organization.Name, organization.InstanceID, organization.State)
-	builder.WriteString(createOrganizationStmt)
+	var (
+		builder              database.StatementBuilder
+		createdAt, updatedAt any = database.DefaultInstruction, database.DefaultInstruction
+	)
 
+	if !organization.CreatedAt.IsZero() {
+		createdAt = organization.CreatedAt
+	}
+	if !organization.UpdatedAt.IsZero() {
+		updatedAt = organization.UpdatedAt
+	}
+
+	builder.WriteString("INSERT INTO ")
+	builder.WriteString(o.qualifiedTableName())
+	builder.WriteString(" (id, name, instance_id, state, created_at, updated_at) VALUES (")
+	builder.WriteArgs(organization.ID, organization.Name, organization.InstanceID, organization.State, createdAt, updatedAt)
+	builder.WriteString(") RETURNING created_at, updated_at")
 	return client.QueryRow(ctx, builder.String(), builder.Args()...).Scan(&organization.CreatedAt, &organization.UpdatedAt)
 }
 
@@ -107,7 +121,9 @@ func (o org) Update(ctx context.Context, client database.QueryExecutor, conditio
 	}
 
 	var builder database.StatementBuilder
-	builder.WriteString(`UPDATE zitadel.organizations SET `)
+	builder.WriteString(`UPDATE `)
+	builder.WriteString(o.qualifiedTableName())
+	builder.WriteString(` SET `)
 	err := database.Changes(changes).Write(&builder)
 	if err != nil {
 		return 0, err
@@ -127,7 +143,8 @@ func (o org) Delete(ctx context.Context, client database.QueryExecutor, conditio
 	}
 
 	var builder database.StatementBuilder
-	builder.WriteString(`DELETE FROM zitadel.organizations`)
+	builder.WriteString(`DELETE FROM `)
+	builder.WriteString(o.qualifiedTableName())
 	writeCondition(&builder, condition)
 
 	return client.Exec(ctx, builder.String(), builder.Args()...)
@@ -147,6 +164,11 @@ func (o org) SetState(state domain.OrgState) database.Change {
 	return database.NewChange(o.StateColumn(), state)
 }
 
+// SetUpdatedAt implements [domain.organizationChanges].
+func (o org) SetUpdatedAt(updatedAt time.Time) database.Change {
+	return database.NewChange(o.UpdatedAtColumn(), updatedAt)
+}
+
 // -------------------------------------------------------------
 // conditions
 // -------------------------------------------------------------

backend/v3/storage/database/repository/project.go

@@ -39,14 +39,22 @@ func (p project) List(ctx context.Context, client database.QueryExecutor, opts .
 	return getMany[domain.Project](ctx, client, builder)
 }
 
-const insertProjectStmt = `INSERT INTO zitadel.projects(
-	instance_id, organization_id, id, name, state, should_assert_role, is_authorization_required, is_project_access_required, used_labeling_setting_owner
-)
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
-RETURNING created_at, updated_at`
+func (p project) Create(ctx context.Context, client database.QueryExecutor, project *domain.Project) error {
+	var (
+		builder              database.StatementBuilder
+		createdAt, updatedAt any = database.DefaultInstruction, database.DefaultInstruction
+	)
+	if !project.CreatedAt.IsZero() {
+		createdAt = project.CreatedAt
+	}
+	if !project.UpdatedAt.IsZero() {
+		updatedAt = project.UpdatedAt
+	}
 
-func (project) Create(ctx context.Context, client database.QueryExecutor, project *domain.Project) error {
-	builder := database.NewStatementBuilder(insertProjectStmt,
+	builder.WriteString("INSERT INTO ")
+	builder.WriteString(p.qualifiedTableName())
+	builder.WriteString("(instance_id, organization_id, id, name, state, should_assert_role, is_authorization_required, is_project_access_required, used_labeling_setting_owner, created_at, updated_at) VALUES (")
+	builder.WriteArgs(
 		project.InstanceID,
 		project.OrganizationID,
 		project.ID,
@@ -56,7 +64,11 @@ func (project) Create(ctx context.Context, client database.QueryExecutor, projec
 		project.IsAuthorizationRequired,
 		project.IsProjectAccessRequired,
 		project.UsedLabelingSettingOwner,
+		createdAt,
+		updatedAt,
 	)
+	builder.WriteString(`) RETURNING created_at, updated_at`)
+
 	return client.QueryRow(ctx, builder.String(), builder.Args()...).
 		Scan(&project.CreatedAt, &project.UpdatedAt)
 }
@@ -141,7 +153,7 @@ func (p project) StateCondition(state domain.ProjectState) database.Condition {
 // -------------------------------------------------------------
 
 func (p project) qualifiedTableName() string {
-	return "zitadel." + p.unqualifiedTableName()
+	return "zitadel.projects"
 }
 
 func (project) unqualifiedTableName() string {

backend/v3/storage/database/repository/project_grant.go

@@ -64,7 +64,9 @@ func (p projectGrant) Create(ctx context.Context, client database.QueryExecutor,
 	// separate statement to add roles to project grant
 	builder := database.NewStatementBuilder(insertProjectGrantRolesStmt, projectGrant.RoleKeys)
 
-	builder.WriteString(`INSERT INTO ` + p.qualifiedTableName() + ` (instance_id, id, project_id, granting_organization_id, granted_organization_id, state, created_at, updated_at) VALUES ( `)
+	builder.WriteString(`INSERT INTO `)
+	builder.WriteString(p.qualifiedTableName())
+	builder.WriteString(` (instance_id, id, project_id, granting_organization_id, granted_organization_id, state, created_at, updated_at) VALUES ( `)
 	builder.WriteArgs(
 		projectGrant.InstanceID,
 		projectGrant.ID,

backend/v3/storage/database/repository/project_role.go

@@ -14,6 +14,10 @@ import (
 
 type projectRole struct{}
 
+func (projectRole) qualifiedTableName() string {
+	return "zitadel.project_roles"
+}
+
 func (p projectRole) Get(ctx context.Context, client database.QueryExecutor, opts ...database.QueryOption) (*domain.ProjectRole, error) {
 	builder, err := p.prepareQuery(opts)
 	if err != nil {
@@ -30,21 +34,32 @@ func (p projectRole) List(ctx context.Context, client database.QueryExecutor, op
 	return getMany[domain.ProjectRole](ctx, client, builder)
 }
 
-const insertProjectRoleStmt = `INSERT INTO zitadel.project_roles(
-	instance_id, organization_id, project_id, key, display_name, role_group
-)
-VALUES ($1, $2, $3, $4, $5, $6)
-RETURNING created_at, updated_at`
-
 func (p projectRole) Create(ctx context.Context, client database.QueryExecutor, role *domain.ProjectRole) error {
-	builder := database.NewStatementBuilder(insertProjectRoleStmt,
+	var (
+		builder              database.StatementBuilder
+		createdAt, updatedAt any = database.DefaultInstruction, database.DefaultInstruction
+	)
+	if !role.CreatedAt.IsZero() {
+		createdAt = role.CreatedAt
+	}
+	if !role.UpdatedAt.IsZero() {
+		updatedAt = role.UpdatedAt
+	}
+
+	builder.WriteString("INSERT INTO ")
+	builder.WriteString(p.qualifiedTableName())
+	builder.WriteString(` (instance_id, organization_id, project_id, key, display_name, role_group, created_at, updated_at) VALUES (`)
+	builder.WriteArgs(
 		role.InstanceID,
 		role.OrganizationID,
 		role.ProjectID,
 		role.Key,
 		role.DisplayName,
 		role.RoleGroup,
+		createdAt,
+		updatedAt,
 	)
+	builder.WriteString(`) RETURNING created_at, updated_at`)
 	return client.QueryRow(ctx, builder.String(), builder.Args()...).
 		Scan(&role.CreatedAt, &role.UpdatedAt)
 }

internal/query/projection/org_relational.go

@@ -2,8 +2,11 @@ package projection
 
 import (
 	"context"
+	"database/sql"
 
-	repoDomain "github.com/zitadel/zitadel/backend/v3/domain"
+	"github.com/zitadel/zitadel/backend/v3/domain"
+	v3_sql "github.com/zitadel/zitadel/backend/v3/storage/database/dialect/sql"
+	"github.com/zitadel/zitadel/backend/v3/storage/database/repository"
 	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
 	"github.com/zitadel/zitadel/internal/repository/instance"
@@ -50,11 +53,6 @@ func (p *orgRelationalProjection) Reducers() []handler.AggregateReducer {
 					Event:  org.OrgRemovedEventType,
 					Reduce: p.reduceOrgRelationalRemoved,
 				},
-				// TODO
-				// {
-				// 	Event:  org.OrgDomainPrimarySetEventType,
-				// 	Reduce: p.reducePrimaryDomainSetRelational,
-				// },
 			},
 		},
 		{
@@ -75,17 +73,22 @@ func (p *orgRelationalProjection) reduceOrgRelationalAdded(event eventstore.Even
 		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-uYq5R", "reduce.wrong.event.type %s", org.OrgAddedEventType)
 	}
 
-	return handler.NewCreateStatement(
-		e,
-		[]handler.Column{
-			handler.NewCol(OrgColumnID, e.Aggregate().ID),
-			handler.NewCol(OrgColumnName, e.Name),
-			handler.NewCol(OrgColumnInstanceID, e.Aggregate().InstanceID),
-			handler.NewCol(State, repoDomain.OrgStateActive),
-			handler.NewCol(CreatedAt, e.CreationDate()),
-			handler.NewCol(UpdatedAt, e.CreationDate()),
-		},
-	), nil
+	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, _ string) error {
+		tx, ok := ex.(*sql.Tx)
+		if !ok {
+			return zerrors.ThrowInvalidArgumentf(nil, "HANDL-K6rs9Y", "reduce.wrong.db.pool %T", ex)
+		}
+
+		repo := repository.OrganizationRepository()
+		return repo.Create(ctx, v3_sql.SQLTx(tx), &domain.Organization{
+			ID:         e.Aggregate().ID,
+			Name:       e.Name,
+			InstanceID: e.Aggregate().InstanceID,
+			State:      domain.OrgStateActive,
+			CreatedAt:  e.CreatedAt(),
+			UpdatedAt:  e.CreatedAt(),
+		})
+	}), nil
 }
 
 func (p *orgRelationalProjection) reduceOrgRelationalChanged(event eventstore.Event) (*handler.Statement, error) {
@@ -96,17 +99,20 @@ func (p *orgRelationalProjection) reduceOrgRelationalChanged(event eventstore.Ev
 	if e.Name == "" {
 		return handler.NewNoOpStatement(e), nil
 	}
-	return handler.NewUpdateStatement(
-		e,
-		[]handler.Column{
-			handler.NewCol(OrgColumnName, e.Name),
-			handler.NewCol(UpdatedAt, e.CreationDate()),
-		},
-		[]handler.Condition{
-			handler.NewCond(OrgColumnID, e.Aggregate().ID),
-			handler.NewCond(OrgColumnInstanceID, e.Aggregate().InstanceID),
-		},
-	), nil
+	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, _ string) error {
+		tx, ok := ex.(*sql.Tx)
+		if !ok {
+			return zerrors.ThrowInvalidArgumentf(nil, "HANDL-ZrVXLa", "reduce.wrong.db.pool %T", ex)
+		}
+
+		repo := repository.OrganizationRepository()
+		_, err := repo.Update(ctx, v3_sql.SQLTx(tx),
+			repo.PrimaryKeyCondition(e.Aggregate().InstanceID, e.Aggregate().ID),
+			repo.SetName(e.Name),
+			repo.SetUpdatedAt(e.CreatedAt()),
+		)
+		return err
+	}), nil
 }
 
 func (p *orgRelationalProjection) reduceOrgRelationalDeactivated(event eventstore.Event) (*handler.Statement, error) {
@@ -115,17 +121,20 @@ func (p *orgRelationalProjection) reduceOrgRelationalDeactivated(event eventstor
 		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-BApK5", "reduce.wrong.event.type %s", org.OrgDeactivatedEventType)
 	}
 
-	return handler.NewUpdateStatement(
-		e,
-		[]handler.Column{
-			handler.NewCol(State, repoDomain.OrgStateInactive),
-			handler.NewCol(UpdatedAt, e.CreationDate()),
-		},
-		[]handler.Condition{
-			handler.NewCond(OrgColumnID, e.Aggregate().ID),
-			handler.NewCond(OrgColumnInstanceID, e.Aggregate().InstanceID),
-		},
-	), nil
+	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, _ string) error {
+		tx, ok := ex.(*sql.Tx)
+		if !ok {
+			return zerrors.ThrowInvalidArgumentf(nil, "HANDL-ro7g4", "reduce.wrong.db.pool %T", ex)
+		}
+
+		repo := repository.OrganizationRepository()
+		_, err := repo.Update(ctx, v3_sql.SQLTx(tx),
+			repo.PrimaryKeyCondition(e.Aggregate().InstanceID, e.Aggregate().ID),
+			repo.SetState(domain.OrgStateInactive),
+			repo.SetUpdatedAt(e.CreatedAt()),
+		)
+		return err
+	}), nil
 }
 
 func (p *orgRelationalProjection) reduceOrgRelationalReactivated(event eventstore.Event) (*handler.Statement, error) {
@@ -133,49 +142,35 @@ func (p *orgRelationalProjection) reduceOrgRelationalReactivated(event eventstor
 	if !ok {
 		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-o38DE", "reduce.wrong.event.type %s", org.OrgReactivatedEventType)
 	}
-	return handler.NewUpdateStatement(
-		e,
-		[]handler.Column{
-			handler.NewCol(State, repoDomain.OrgStateActive),
-			handler.NewCol(UpdatedAt, e.CreationDate()),
-		},
-		[]handler.Condition{
-			handler.NewCond(OrgColumnID, e.Aggregate().ID),
-			handler.NewCond(OrgColumnInstanceID, e.Aggregate().InstanceID),
-		},
-	), nil
-}
+	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, _ string) error {
+		tx, ok := ex.(*sql.Tx)
+		if !ok {
+			return zerrors.ThrowInvalidArgumentf(nil, "HANDL-5snmW", "reduce.wrong.db.pool %T", ex)
+		}
 
-// TODO
-// func (p *orgRelationalProjection) reducePrimaryDomainSetRelational(event eventstore.Event) (*handler.Statement, error) {
-// 	e, ok := event.(*org.DomainPrimarySetEvent)
-// 	if !ok {
-// 		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-3Tbkt", "reduce.wrong.event.type %s", org.OrgDomainPrimarySetEventType)
-// 	}
-// 	return handler.NewUpdateStatement(
-// 		e,
-// 		[]handler.Column{
-// 			handler.NewCol(OrgColumnChangeDate, e.CreationDate()),
-// 			handler.NewCol(OrgColumnSequence, e.Sequence()),
-// 			handler.NewCol(OrgColumnDomain, e.Domain),
-// 		},
-// 		[]handler.Condition{
-// 			handler.NewCond(OrgColumnID, e.Aggregate().ID),
-// 			handler.NewCond(OrgColumnInstanceID, e.Aggregate().InstanceID),
-// 		},
-// 	), nil
-// }
+		repo := repository.OrganizationRepository()
+		_, err := repo.Update(ctx, v3_sql.SQLTx(tx),
+			repo.PrimaryKeyCondition(e.Aggregate().InstanceID, e.Aggregate().ID),
+			repo.SetState(domain.OrgStateActive),
+			repo.SetUpdatedAt(e.CreatedAt()),
+		)
+		return err
+	}), nil
+}
 
 func (p *orgRelationalProjection) reduceOrgRelationalRemoved(event eventstore.Event) (*handler.Statement, error) {
 	e, ok := event.(*org.OrgRemovedEvent)
 	if !ok {
-		return nil, zerrors.ThrowInvalidArgumentf(nil, "PROJE-DGm9g", "reduce.wrong.event.type %s", org.OrgRemovedEventType)
+		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-DGm9g", "reduce.wrong.event.type %s", org.OrgRemovedEventType)
 	}
-	return handler.NewDeleteStatement(
-		e,
-		[]handler.Condition{
-			handler.NewCond(OrgColumnID, e.Aggregate().ID),
-			handler.NewCond(OrgColumnInstanceID, e.Aggregate().InstanceID),
-		},
-	), nil
+	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, _ string) error {
+		tx, ok := ex.(*sql.Tx)
+		if !ok {
+			return zerrors.ThrowInvalidArgumentf(nil, "HANDL-6y7hJ", "reduce.wrong.db.pool %T", ex)
+		}
+
+		repo := repository.OrganizationRepository()
+		_, err := repo.Delete(ctx, v3_sql.SQLTx(tx), repo.PrimaryKeyCondition(e.Aggregate().InstanceID, e.Aggregate().ID))
+		return err
+	}), nil
 }