feat(db): adding org table to relational model (#10066)

# Which Problems Are Solved

As an outcome of [this
issue](https://github.com/zitadel/zitadel/issues/9599) we want to
implement relational tables in Zitadel. For that we use new tables as a
successor of the current tables used by Zitadel in `projections`, `auth`
and `admin` schemas. The new logic is based on [this
proposal](https://github.com/zitadel/zitadel/pull/9870). This issue does
not contain the switch from CQRS to the new tables. This is change will
be implemented in a later stage.

We focus on the most critical tables which is user authentication.

We need a table to manage organizations. 

### organization fields

The following fields must be managed in this table:

- `id`
- `instance_id`
- `name`
- `state` enum (active, inactive)
- `created_at`
- `updated_at`
- `deleted_at`

DISCUSS: should we add a `primary_domain` to this table so that we do
not have to join on domains to return a simple org?

We must ensure the unique constraints for this table matches the current
commands.

### organization repository

The repository must provide the following functions:

Manipulations:
- create
  - `instance_id`
  - `name`
- update
  - `name`
- delete

Queries:
- get returns single organization matching the criteria and pagination,
should return error if multiple were found
- list returns list of organizations matching the criteria, pagination

Criteria are the following:
- by id
- by name

pagination:
- by created_at
- by updated_at
- by name

### organization events

The following events must be applied on the table using a projection
(`internal/query/projection`)

- `org.added` results in create
- `org.changed` sets the `name` field
- `org.deactivated` sets the `state` field
- `org.reactivated` sets the `state` field
- `org.removed` sets the `deleted_at` field
- if answer is yes to discussion: `org.domain.primary.set` sets the
`primary_domain` field
- `instance.removed` sets the the `deleted_at` field if not already set

### acceptance criteria

- [x] migration is implemented and gets executed
- [x] domain interfaces are implemented and documented for service layer
- [x] repository is implemented and implements domain interface
- [x] testing
  - [x] the repository methods
  - [x] events get reduced correctly
  - [x] unique constraints
# Additional Context

Replace this example with links to related issues, discussions, discord
threads, or other sources with more context.
Use the Closing #issue syntax for issues that are resolved with this PR.
- Closes #https://github.com/zitadel/zitadel/issues/9936

---------

Co-authored-by: adlerhurst <27845747+adlerhurst@users.noreply.github.com>

internal/query/projection/org_relational.go

@@ -0,0 +1,185 @@
+package projection
+
+import (
+	"context"
+
+	repoDomain "github.com/zitadel/zitadel/backend/v3/domain"
+	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
+	"github.com/zitadel/zitadel/internal/repository/instance"
+	"github.com/zitadel/zitadel/internal/repository/org"
+	"github.com/zitadel/zitadel/internal/zerrors"
+)
+
+const (
+	OrgRelationProjectionTable = "zitadel.organizations"
+)
+
+type orgRelationalProjection struct{}
+
+func (*orgRelationalProjection) Name() string {
+	return OrgRelationProjectionTable
+}
+
+func newOrgRelationalProjection(ctx context.Context, config handler.Config) *handler.Handler {
+	return handler.NewHandler(ctx, &config, new(orgRelationalProjection))
+}
+
+func (p *orgRelationalProjection) Reducers() []handler.AggregateReducer {
+	return []handler.AggregateReducer{
+		{
+			Aggregate: org.AggregateType,
+			EventReducers: []handler.EventReducer{
+				{
+					Event:  org.OrgAddedEventType,
+					Reduce: p.reduceOrgRelationalAdded,
+				},
+				{
+					Event:  org.OrgChangedEventType,
+					Reduce: p.reduceOrgRelationalChanged,
+				},
+				{
+					Event:  org.OrgDeactivatedEventType,
+					Reduce: p.reduceOrgRelationalDeactivated,
+				},
+				{
+					Event:  org.OrgReactivatedEventType,
+					Reduce: p.reduceOrgRelationalReactivated,
+				},
+				{
+					Event:  org.OrgRemovedEventType,
+					Reduce: p.reduceOrgRelationalRemoved,
+				},
+				// TODO
+				// {
+				// 	Event:  org.OrgDomainPrimarySetEventType,
+				// 	Reduce: p.reducePrimaryDomainSetRelational,
+				// },
+			},
+		},
+		{
+			Aggregate: instance.AggregateType,
+			EventReducers: []handler.EventReducer{
+				{
+					Event:  instance.InstanceRemovedEventType,
+					Reduce: reduceInstanceRemovedHelper(OrgColumnInstanceID),
+				},
+			},
+		},
+	}
+}
+
+func (p *orgRelationalProjection) reduceOrgRelationalAdded(event eventstore.Event) (*handler.Statement, error) {
+	e, ok := event.(*org.OrgAddedEvent)
+	if !ok {
+		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.String()),
+			handler.NewCol(CreatedAt, e.CreationDate()),
+			handler.NewCol(UpdatedAt, e.CreationDate()),
+		},
+	), nil
+}
+
+func (p *orgRelationalProjection) reduceOrgRelationalChanged(event eventstore.Event) (*handler.Statement, error) {
+	e, ok := event.(*org.OrgChangedEvent)
+	if !ok {
+		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-Bg9om", "reduce.wrong.event.type %s", org.OrgChangedEventType)
+	}
+	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
+}
+
+func (p *orgRelationalProjection) reduceOrgRelationalDeactivated(event eventstore.Event) (*handler.Statement, error) {
+	e, ok := event.(*org.OrgDeactivatedEvent)
+	if !ok {
+		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.String()),
+			handler.NewCol(UpdatedAt, e.CreationDate()),
+		},
+		[]handler.Condition{
+			handler.NewCond(OrgColumnID, e.Aggregate().ID),
+			handler.NewCond(OrgColumnInstanceID, e.Aggregate().InstanceID),
+		},
+	), nil
+}
+
+func (p *orgRelationalProjection) reduceOrgRelationalReactivated(event eventstore.Event) (*handler.Statement, error) {
+	e, ok := event.(*org.OrgReactivatedEvent)
+	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.String()),
+			handler.NewCol(UpdatedAt, e.CreationDate()),
+		},
+		[]handler.Condition{
+			handler.NewCond(OrgColumnID, e.Aggregate().ID),
+			handler.NewCond(OrgColumnInstanceID, e.Aggregate().InstanceID),
+		},
+	), nil
+}
+
+// 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
+// }
+
+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 handler.NewUpdateStatement(
+		e,
+		[]handler.Column{
+			handler.NewCol(UpdatedAt, e.CreationDate()),
+			handler.NewCol(DeletedAt, e.CreationDate()),
+		},
+		[]handler.Condition{
+			handler.NewCond(OrgColumnID, e.Aggregate().ID),
+			handler.NewCond(OrgColumnInstanceID, e.Aggregate().InstanceID),
+		},
+	), nil
+}

internal/query/projection/projection.go

@@ -61,6 +61,7 @@ var (
 	UserAuthMethodProjection            *handler.Handler
 	InstanceProjection                  *handler.Handler
 	InstanceRelationalProjection        *handler.Handler
+	OrganizationRelationalProjection    *handler.Handler
 	SecretGeneratorProjection           *handler.Handler
 	SMTPConfigProjection                *handler.Handler
 	SMSConfigProjection                 *handler.Handler
@@ -157,7 +158,8 @@ func Create(ctx context.Context, sqlClient *database.DB, es handler.EventStore,
 	UserMetadataProjection = newUserMetadataProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["user_metadata"]))
 	UserAuthMethodProjection = newUserAuthMethodProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["user_auth_method"]))
 	InstanceProjection = newInstanceProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["instances"]))
-	InstanceRelationalProjection = newInstanceRelationalProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["instances_relational"]))
+	InstanceRelationalProjection = newInstanceRelationalProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["organizations_relational"]))
+	OrganizationRelationalProjection = newOrgRelationalProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["instances_relational"]))
 	SecretGeneratorProjection = newSecretGeneratorProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["secret_generators"]))
 	SMTPConfigProjection = newSMTPConfigProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["smtp_configs"]))
 	SMSConfigProjection = newSMSConfigProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["sms_config"]))
@@ -337,6 +339,7 @@ func newProjectionsList() {
 		UserAuthMethodProjection,
 		InstanceProjection,
 		InstanceRelationalProjection,
+		OrganizationRelationalProjection,
 		SecretGeneratorProjection,
 		SMTPConfigProjection,
 		SMSConfigProjection,

internal/query/projection/relational_common.go

@@ -1,6 +1,7 @@
 package projection
 
 const (
+	State     = "state"
 	CreatedAt = "created_at"
 	UpdatedAt = "updated_at"
 	DeletedAt = "deleted_at"