feat: test event reduction to relational tables (#11718)

# Which Problems Are Solved

This PR provides an example for testing DB reduction to relational
tables without exploiting integration tests.

# How the Problems Are Solved

- Set up a disposable embedded Postgres DB used to store the results of
the calls to the reducers and verify such results
- Add utility method to return transactions from the existing DB pool.
These transactions can be used by the single test cases to insert and
read data from DB
- Add a test to verify that the handler contains the expected reducers
(checking also that they are on the right aggregate type)
- Add tests to verify the single reduce functions. A parent test creates
the required objects to satisfy the foreign key constraints.

# Additional Changes

  - Typo fixed in `postgres.DecodeConfig()`
  - Discard unneeded `projectionName` parameter for `Executor` function
 
# Additional Context

- Closes #11430

backend/v3/storage/database/database.go

@@ -2,6 +2,7 @@ package database
 
 import (
 	"context"
+	"database/sql"
 )
 
 // Pool is a connection pool. e.g. pgxpool
@@ -20,6 +21,9 @@ type PoolTest interface {
 	Pool
 	// MigrateTest is the same as [Migrator] but executes the migrations multiple times instead of only once.
 	MigrateTest(ctx context.Context) error
+	// RawDB returns an *[sql.DB] handle backed by the same database as the [PoolTest].
+	// The returned handle is intended for use in tests, and the caller is responsible for closing it when done.
+	RawDB() *sql.DB
 }
 
 // Connection is a single database connection which can be released back to the pool.

backend/v3/storage/database/dialect/postgres/config.go

@@ -81,9 +81,7 @@ func DecodeConfig(input any) (database.Connector, error) {
 		if err = decoder.Decode(c); err != nil {
 			return nil, err
 		}
-		return &Config{
-			Config: &pgxpool.Config{},
-		}, nil
+		return connector, nil
 	}
 	return nil, errors.New("invalid configuration")
 }

backend/v3/storage/database/dialect/postgres/pool.go

@@ -2,8 +2,10 @@ package postgres
 
 import (
 	"context"
+	"database/sql"
 
 	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/jackc/pgx/v5/stdlib"
 
 	"github.com/zitadel/zitadel/backend/v3/storage/database"
 	"github.com/zitadel/zitadel/backend/v3/storage/database/dialect/postgres/migration"
@@ -13,7 +15,13 @@ type pgxPool struct {
 	*pgxpool.Pool
 }
 
+// RawDB implements [database.PoolTest].
+func (p *pgxPool) RawDB() *sql.DB {
+	return stdlib.OpenDBFromPool(p.Pool)
+}
+
 var _ database.Pool = (*pgxPool)(nil)
+var _ database.PoolTest = (*pgxPool)(nil)
 
 func PGxPool(pool *pgxpool.Pool) *pgxPool {
 	return &pgxPool{

backend/v3/storage/database/events_testing/instance_domain_test.go

@@ -1,310 +0,0 @@
-//go:build integration
-
-package events_test
-
-import (
-	"testing"
-	"time"
-
-	"github.com/brianvoe/gofakeit/v6"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"github.com/zitadel/zitadel/backend/v3/domain"
-	"github.com/zitadel/zitadel/backend/v3/storage/database"
-	"github.com/zitadel/zitadel/backend/v3/storage/database/repository"
-	"github.com/zitadel/zitadel/internal/integration"
-	v2beta "github.com/zitadel/zitadel/pkg/grpc/instance/v2beta"
-	"github.com/zitadel/zitadel/pkg/grpc/system"
-)
-
-func TestServer_TestInstanceDomainReduces(t *testing.T) {
-	instance := integration.NewInstance(CTX)
-
-	instanceRepo := repository.InstanceRepository()
-	instanceDomainRepo := repository.InstanceDomainRepository()
-
-	t.Cleanup(func() {
-		_, err := instance.Client.InstanceV2Beta.DeleteInstance(CTX, &v2beta.DeleteInstanceRequest{
-			InstanceId: instance.Instance.Id,
-		})
-		if err != nil {
-			t.Logf("Failed to delete instance on cleanup: %v", err)
-		}
-	})
-
-	// Wait for instance to be created
-	retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
-	assert.EventuallyWithT(t, func(t *assert.CollectT) {
-		_, err := instanceRepo.Get(CTX, pool,
-			database.WithCondition(instanceRepo.IDCondition(instance.ID())),
-		)
-		assert.NoError(t, err)
-	}, retryDuration, tick)
-
-	t.Run("test instance custom domain add reduces", func(t *testing.T) {
-		// Add a domain to the instance
-		domainName := gofakeit.DomainName()
-		beforeAdd := time.Now()
-		_, err := instance.Client.InstanceV2Beta.AddCustomDomain(CTX, &v2beta.AddCustomDomainRequest{
-			InstanceId: instance.Instance.Id,
-			Domain:     domainName,
-		})
-		require.NoError(t, err)
-		afterAdd := time.Now()
-
-		t.Cleanup(func() {
-			_, err := instance.Client.InstanceV2Beta.RemoveCustomDomain(CTX, &v2beta.RemoveCustomDomainRequest{
-				InstanceId: instance.Instance.Id,
-				Domain:     domainName,
-			})
-			if err != nil {
-				t.Logf("Failed to delete instance domain on cleanup: %v", err)
-			}
-		})
-
-		// Test that domain add reduces
-		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
-		assert.EventuallyWithT(t, func(t *assert.CollectT) {
-			domain, err := instanceDomainRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						instanceDomainRepo.InstanceIDCondition(instance.Instance.Id),
-						instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
-						instanceDomainRepo.TypeCondition(domain.DomainTypeCustom),
-					),
-				),
-			)
-			require.NoError(t, err)
-			// event instance.domain.added
-			assert.Equal(t, domainName, domain.Domain)
-			assert.Equal(t, instance.Instance.Id, domain.InstanceID)
-			assert.False(t, *domain.IsPrimary)
-			assert.WithinRange(t, domain.CreatedAt, beforeAdd, afterAdd)
-			assert.WithinRange(t, domain.UpdatedAt, beforeAdd, afterAdd)
-		}, retryDuration, tick)
-	})
-
-	t.Run("test instance custom domain set primary reduces", func(t *testing.T) {
-		// Add a domain to the instance
-		domainName := gofakeit.DomainName()
-		_, err := instance.Client.InstanceV2Beta.AddCustomDomain(CTX, &v2beta.AddCustomDomainRequest{
-			InstanceId: instance.Instance.Id,
-			Domain:     domainName,
-		})
-		require.NoError(t, err)
-
-		t.Cleanup(func() {
-			// first we change the primary domain to something else
-			domain, err := instanceDomainRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						instanceDomainRepo.InstanceIDCondition(instance.Instance.Id),
-						instanceDomainRepo.TypeCondition(domain.DomainTypeCustom),
-						instanceDomainRepo.IsPrimaryCondition(false),
-					),
-				),
-				database.WithLimit(1),
-			)
-			require.NoError(t, err)
-			_, err = SystemClient.SetPrimaryDomain(CTX, &system.SetPrimaryDomainRequest{
-				InstanceId: instance.Instance.Id,
-				Domain:     domain.Domain,
-			})
-			require.NoError(t, err)
-
-			_, err = instance.Client.InstanceV2Beta.RemoveCustomDomain(CTX, &v2beta.RemoveCustomDomainRequest{
-				InstanceId: instance.Instance.Id,
-				Domain:     domainName,
-			})
-			if err != nil {
-				t.Logf("Failed to delete instance domain on cleanup: %v", err)
-			}
-		})
-
-		// Wait for domain to be created
-		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
-		assert.EventuallyWithT(t, func(t *assert.CollectT) {
-			domain, err := instanceDomainRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						instanceDomainRepo.InstanceIDCondition(instance.Instance.Id),
-						instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
-						instanceDomainRepo.TypeCondition(domain.DomainTypeCustom),
-					),
-				),
-			)
-			require.NoError(t, err)
-			require.False(t, *domain.IsPrimary)
-			assert.Equal(t, domainName, domain.Domain)
-		}, retryDuration, tick)
-
-		// Set domain as primary
-		beforeSetPrimary := time.Now()
-		_, err = SystemClient.SetPrimaryDomain(CTX, &system.SetPrimaryDomainRequest{
-			InstanceId: instance.Instance.Id,
-			Domain:     domainName,
-		})
-		require.NoError(t, err)
-		afterSetPrimary := time.Now()
-
-		// Test that set primary reduces
-		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
-		assert.EventuallyWithT(t, func(t *assert.CollectT) {
-			domain, err := instanceDomainRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						instanceDomainRepo.InstanceIDCondition(instance.Instance.Id),
-						instanceDomainRepo.IsPrimaryCondition(true),
-						instanceDomainRepo.TypeCondition(domain.DomainTypeCustom),
-					),
-				),
-			)
-			require.NoError(t, err)
-			// event instance.domain.primary.set
-			assert.Equal(t, domainName, domain.Domain)
-			assert.True(t, *domain.IsPrimary)
-			assert.WithinRange(t, domain.UpdatedAt, beforeSetPrimary, afterSetPrimary)
-		}, retryDuration, tick)
-	})
-
-	t.Run("test instance custom domain remove reduces", func(t *testing.T) {
-		// Add a domain to the instance
-		domainName := gofakeit.DomainName()
-		_, err := instance.Client.InstanceV2Beta.AddCustomDomain(CTX, &v2beta.AddCustomDomainRequest{
-			InstanceId: instance.Instance.Id,
-			Domain:     domainName,
-		})
-		require.NoError(t, err)
-
-		// Wait for domain to be created and verify it exists
-		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
-		assert.EventuallyWithT(t, func(t *assert.CollectT) {
-			_, err := instanceDomainRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						instanceDomainRepo.InstanceIDCondition(instance.Instance.Id),
-						instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
-						instanceDomainRepo.TypeCondition(domain.DomainTypeCustom),
-					),
-				),
-			)
-			require.NoError(t, err)
-		}, retryDuration, tick)
-
-		// Remove the domain
-		_, err = instance.Client.InstanceV2Beta.RemoveCustomDomain(CTX, &v2beta.RemoveCustomDomainRequest{
-			InstanceId: instance.Instance.Id,
-			Domain:     domainName,
-		})
-		require.NoError(t, err)
-
-		// Test that domain remove reduces
-		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
-		assert.EventuallyWithT(t, func(t *assert.CollectT) {
-			domain, err := instanceDomainRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						instanceDomainRepo.InstanceIDCondition(instance.Instance.Id),
-						instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
-						instanceDomainRepo.TypeCondition(domain.DomainTypeCustom),
-					),
-				),
-			)
-			// event instance.domain.removed
-			assert.Nil(t, domain)
-			require.ErrorIs(t, err, new(database.NoRowFoundError))
-		}, retryDuration, tick)
-	})
-
-	t.Run("test instance trusted domain add reduces", func(t *testing.T) {
-		// Add a domain to the instance
-		domainName := gofakeit.DomainName()
-		beforeAdd := time.Now()
-		_, err := instance.Client.InstanceV2Beta.AddTrustedDomain(CTX, &v2beta.AddTrustedDomainRequest{
-			InstanceId: instance.Instance.Id,
-			Domain:     domainName,
-		})
-		require.NoError(t, err)
-		afterAdd := time.Now()
-
-		t.Cleanup(func() {
-			_, err := instance.Client.InstanceV2Beta.RemoveTrustedDomain(CTX, &v2beta.RemoveTrustedDomainRequest{
-				InstanceId: instance.Instance.Id,
-				Domain:     domainName,
-			})
-			if err != nil {
-				t.Logf("Failed to delete instance domain on cleanup: %v", err)
-			}
-		})
-
-		// Test that domain add reduces
-		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
-		assert.EventuallyWithT(t, func(t *assert.CollectT) {
-			domain, err := instanceDomainRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						instanceDomainRepo.InstanceIDCondition(instance.Instance.Id),
-						instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
-						instanceDomainRepo.TypeCondition(domain.DomainTypeTrusted),
-					),
-				),
-			)
-			require.NoError(t, err)
-			// event instance.domain.added
-			assert.Equal(t, domainName, domain.Domain)
-			assert.Equal(t, instance.Instance.Id, domain.InstanceID)
-			assert.WithinRange(t, domain.CreatedAt, beforeAdd, afterAdd)
-			assert.WithinRange(t, domain.UpdatedAt, beforeAdd, afterAdd)
-		}, retryDuration, tick)
-	})
-
-	t.Run("test instance trusted domain remove reduces", func(t *testing.T) {
-		// Add a domain to the instance
-		domainName := gofakeit.DomainName()
-		_, err := instance.Client.InstanceV2Beta.AddTrustedDomain(CTX, &v2beta.AddTrustedDomainRequest{
-			InstanceId: instance.Instance.Id,
-			Domain:     domainName,
-		})
-		require.NoError(t, err)
-
-		// Wait for domain to be created and verify it exists
-		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
-		assert.EventuallyWithT(t, func(t *assert.CollectT) {
-			_, err := instanceDomainRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						instanceDomainRepo.InstanceIDCondition(instance.Instance.Id),
-						instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
-						instanceDomainRepo.TypeCondition(domain.DomainTypeTrusted),
-					),
-				),
-			)
-			require.NoError(t, err)
-		}, retryDuration, tick)
-
-		// Remove the domain
-		_, err = instance.Client.InstanceV2Beta.RemoveTrustedDomain(CTX, &v2beta.RemoveTrustedDomainRequest{
-			InstanceId: instance.Instance.Id,
-			Domain:     domainName,
-		})
-		require.NoError(t, err)
-
-		// Test that domain remove reduces
-		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
-		assert.EventuallyWithT(t, func(t *assert.CollectT) {
-			domain, err := instanceDomainRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						instanceDomainRepo.InstanceIDCondition(instance.Instance.Id),
-						instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
-						instanceDomainRepo.TypeCondition(domain.DomainTypeTrusted),
-					),
-				),
-			)
-			// event instance.domain.removed
-			assert.Nil(t, domain)
-			require.ErrorIs(t, err, new(database.NoRowFoundError))
-		}, retryDuration, tick)
-	})
-}

internal/query/projection/event_testing_relational_test.go

@@ -0,0 +1,116 @@
+package projection
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log"
+	"os"
+	"slices"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+	"github.com/zitadel/zitadel/backend/v3/storage/database/dialect/postgres/embedded"
+	v3_sql "github.com/zitadel/zitadel/backend/v3/storage/database/dialect/sql"
+	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
+)
+
+var (
+	pool  database.PoolTest
+	rawDB *sql.DB
+)
+
+func TestMain(m *testing.M) {
+	os.Exit(runTests(m))
+}
+
+func runTests(m *testing.M) int {
+	var stop func()
+	var err error
+	ctx := context.Background()
+	pool, stop, err = newEmbeddedDB(ctx)
+	if err != nil {
+		log.Printf("error with embedded postgres database: %v", err)
+		return 1
+	}
+	defer stop()
+
+	rawDB = pool.RawDB()
+	defer rawDB.Close()
+	defer pool.Close(ctx)
+
+	return m.Run()
+}
+
+func newEmbeddedDB(ctx context.Context) (pool database.PoolTest, stop func(), err error) {
+	connector, stop, err := embedded.StartEmbedded()
+	if err != nil {
+		return nil, nil, fmt.Errorf("unable to start embedded postgres: %w", err)
+	}
+
+	dummyPool, err := connector.Connect(ctx)
+	if err != nil {
+		return nil, stop, fmt.Errorf("unable to connect to embedded postgres: %w", err)
+	}
+
+	pool, ok := dummyPool.(database.PoolTest)
+	if !ok {
+		return nil, stop, fmt.Errorf("expecting database.PoolTest, got %T", dummyPool)
+	}
+
+	err = pool.MigrateTest(ctx)
+	if err != nil {
+		return nil, stop, fmt.Errorf("unable to migrate database: %w", err)
+	}
+
+	return pool, stop, err
+}
+
+func getTransactions(t *testing.T) (rawTx *sql.Tx, v3SQLTx *v3_sql.Transaction) {
+	rawTx, err := rawDB.Begin()
+	require.NoError(t, err)
+	v3SQLTx = v3_sql.SQLTx(rawTx)
+	return
+}
+
+// callReduce on the input projection tests that:
+//
+//  1. the aggregate type of the input event exists
+//  2. the event type of the input event exists for that aggregate type
+//  3. the reducer function for the input event is not nil
+//  4. the reducer function for the input event executes without errors
+//
+// Returns false if any of the above checks fails
+func callReduce(t *testing.T, ctx context.Context, tx *sql.Tx, projection handler.Projection, event eventstore.Event) bool {
+	reducers := projection.Reducers()
+	aggregateReducerIdx := slices.IndexFunc(reducers, func(r handler.AggregateReducer) bool {
+		return r.Aggregate == event.Aggregate().Type
+	})
+	if !assert.Greater(t, aggregateReducerIdx, -1) {
+		return false
+	}
+
+	eventReducerIdx := slices.IndexFunc(reducers[aggregateReducerIdx].EventReducers, func(er handler.EventReducer) bool {
+		return er.Event == event.Type()
+	})
+	if !assert.Greater(t, eventReducerIdx, -1) {
+		return false
+	}
+
+	reduceFn := reducers[aggregateReducerIdx].EventReducers[eventReducerIdx].Reduce
+	if !assert.NotNil(t, reduceFn) {
+		return false
+	}
+
+	stmt, err := reduceFn(event)
+	if !assert.NoError(t, err) {
+		return false
+	}
+
+	err = stmt.Execute(ctx, tx, "")
+	return assert.NoError(t, err)
+}

internal/query/projection/instance_domain_relational.go

@@ -61,7 +61,7 @@ func (p *instanceDomainRelationalProjection) reduceCustomDomainAdded(event event
 	if !ok {
 		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-DU0xF", "reduce.wrong.event.type %s", instance.InstanceDomainAddedEventType)
 	}
-	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, projectionName string) error {
+	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-bXCa6", "reduce.wrong.db.pool %T", ex)
@@ -83,7 +83,7 @@ func (p *instanceDomainRelationalProjection) reduceDomainPrimarySet(event events
 	if !ok {
 		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-TdEWA", "reduce.wrong.event.type %s", instance.InstanceDomainPrimarySetEventType)
 	}
-	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, projectionName string) error {
+	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-QnjHo", "reduce.wrong.db.pool %T", ex)
@@ -108,7 +108,7 @@ func (p *instanceDomainRelationalProjection) reduceCustomDomainRemoved(event eve
 	if !ok {
 		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-Hhcdl", "reduce.wrong.event.type %s", instance.InstanceDomainRemovedEventType)
 	}
-	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, projectionName string) error {
+	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-58ghE", "reduce.wrong.db.pool %T", ex)
@@ -130,7 +130,7 @@ func (p *instanceDomainRelationalProjection) reduceTrustedDomainAdded(event even
 	if !ok {
 		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-svHDh", "reduce.wrong.event.type %s", instance.TrustedDomainAddedEventType)
 	}
-	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, projectionName string) error {
+	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-gx7tQ", "reduce.wrong.db.pool %T", ex)
@@ -150,7 +150,7 @@ func (p *instanceDomainRelationalProjection) reduceTrustedDomainRemoved(event ev
 	if !ok {
 		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-4K74E", "reduce.wrong.event.type %s", instance.TrustedDomainRemovedEventType)
 	}
-	return handler.NewStatement(e, func(ctx context.Context, ex handler.Executer, projectionName string) error {
+	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-D68ap", "reduce.wrong.db.pool %T", ex)

internal/query/projection/instance_domain_relational_test.go

@@ -0,0 +1,204 @@
+package projection
+
+import (
+	"testing"
+	"time"
+
+	"github.com/brianvoe/gofakeit/v6"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/zitadel/zitadel/backend/v3/domain"
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+	"github.com/zitadel/zitadel/backend/v3/storage/database/repository"
+	"github.com/zitadel/zitadel/internal/repository/instance"
+)
+
+func TestInstanceDomainReducers(t *testing.T) {
+	handler := &instanceDomainRelationalProjection{}
+	rawTx, tx := getTransactions(t)
+
+	t.Cleanup(func() {
+		require.NoError(t, rawTx.Rollback())
+	})
+	ctx := t.Context()
+
+	instanceDomainRepo := repository.InstanceDomainRepository()
+	instanceRepo := repository.InstanceRepository()
+	err := instanceRepo.Create(ctx, tx, &domain.Instance{
+		ID:              "123",
+		Name:            "my instance",
+		DefaultOrgID:    gofakeit.UUID(),
+		IAMProjectID:    gofakeit.UUID(),
+		ConsoleClientID: gofakeit.UUID(),
+		ConsoleAppID:    gofakeit.UUID(),
+		DefaultLanguage: gofakeit.LanguageAbbreviation(),
+		CreatedAt:       time.Now(),
+		UpdatedAt:       time.Now(),
+	})
+	require.NoError(t, err)
+
+	t.Run("'custom domain added' reducer should create custom domain", func(t *testing.T) {
+		// Given
+		domainName := "test-domain.com"
+		domainAddedEvt := instance.NewDomainAddedEvent(ctx, &instance.NewAggregate("123").Aggregate, domainName, false)
+
+		// Test
+		res := callReduce(t, ctx, rawTx, handler, domainAddedEvt)
+		require.True(t, res)
+
+		// Verify
+		domain, err := instanceDomainRepo.Get(ctx, tx,
+			database.WithCondition(
+				database.And(
+					instanceDomainRepo.InstanceIDCondition("123"),
+					instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
+					instanceDomainRepo.TypeCondition(domain.DomainTypeCustom),
+				),
+			),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, domainName, domain.Domain)
+		assert.Equal(t, "123", domain.InstanceID)
+
+		require.NotNil(t, domain.IsPrimary)
+		assert.False(t, *domain.IsPrimary)
+
+		assert.NotZero(t, domain.CreatedAt)
+		assert.NotZero(t, domain.UpdatedAt)
+	})
+
+	t.Run("'primary domain set' reducer should update existing domain", func(t *testing.T) {
+		// Given
+		domainName := "test-domain-primary-set.com"
+		err := instanceDomainRepo.Add(ctx, tx, &domain.AddInstanceDomain{
+			InstanceID:  "123",
+			Domain:      domainName,
+			Type:        domain.DomainTypeCustom,
+			CreatedAt:   time.Now(),
+			UpdatedAt:   time.Now(),
+			IsPrimary:   new(bool),
+			IsGenerated: new(bool),
+		})
+		require.NoError(t, err)
+
+		domainPrimarySetEvt := instance.NewDomainPrimarySetEvent(ctx, &instance.NewAggregate("123").Aggregate, domainName)
+
+		// Test
+		res := callReduce(t, ctx, rawTx, handler, domainPrimarySetEvt)
+		require.True(t, res)
+
+		// Verify
+		domain, err := instanceDomainRepo.Get(ctx, tx,
+			database.WithCondition(
+				database.And(
+					instanceDomainRepo.InstanceIDCondition("123"),
+					instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
+					instanceDomainRepo.TypeCondition(domain.DomainTypeCustom),
+					instanceDomainRepo.IsPrimaryCondition(true),
+				),
+			),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, domainName, domain.Domain)
+		assert.Equal(t, "123", domain.InstanceID)
+
+		require.NotNil(t, domain.IsPrimary)
+		assert.True(t, *domain.IsPrimary)
+
+		assert.NotZero(t, domain.CreatedAt)
+		assert.NotZero(t, domain.UpdatedAt)
+	})
+
+	t.Run("'custom domain removed' reducer should remove custom domain", func(t *testing.T) {
+		// Given
+		domainName := "test-domain-removed.com"
+		err := instanceDomainRepo.Add(ctx, tx, &domain.AddInstanceDomain{
+			InstanceID:  "123",
+			Domain:      domainName,
+			Type:        domain.DomainTypeCustom,
+			CreatedAt:   time.Now(),
+			UpdatedAt:   time.Now(),
+			IsPrimary:   new(bool),
+			IsGenerated: new(bool),
+		})
+		require.NoError(t, err)
+
+		domainRemovedEvt := instance.NewDomainRemovedEvent(ctx, &instance.NewAggregate("123").Aggregate, domainName)
+
+		// Test
+		res := callReduce(t, ctx, rawTx, handler, domainRemovedEvt)
+		require.True(t, res)
+
+		// Verify
+		_, err = instanceDomainRepo.Get(ctx, tx,
+			database.WithCondition(
+				database.And(
+					instanceDomainRepo.InstanceIDCondition("123"),
+					instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
+					instanceDomainRepo.TypeCondition(domain.DomainTypeCustom),
+				),
+			),
+		)
+		require.Error(t, err)
+		assert.ErrorIs(t, err, &database.NoRowFoundError{})
+	})
+
+	t.Run("'trusted domain added' reducer should create trusted domain", func(t *testing.T) {
+		// Given
+		domainName := "trusted-test-domain.com"
+		trustedDomainAddedEvt := instance.NewTrustedDomainAddedEvent(ctx, &instance.NewAggregate("123").Aggregate, domainName)
+
+		// Test
+		res := callReduce(t, ctx, rawTx, handler, trustedDomainAddedEvt)
+		require.True(t, res)
+
+		// Verify
+		domain, err := instanceDomainRepo.Get(ctx, tx,
+			database.WithCondition(
+				database.And(
+					instanceDomainRepo.InstanceIDCondition("123"),
+					instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
+					instanceDomainRepo.TypeCondition(domain.DomainTypeTrusted),
+				),
+			),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, domainName, domain.Domain)
+		assert.Equal(t, "123", domain.InstanceID)
+		assert.NotZero(t, domain.CreatedAt)
+		assert.NotZero(t, domain.UpdatedAt)
+	})
+
+	t.Run("'trusted domain removed' reducer should remove trusted domain", func(t *testing.T) {
+		// Given
+		domainName := "test-trusted-domain-removed.com"
+		err := instanceDomainRepo.Add(ctx, tx, &domain.AddInstanceDomain{
+			InstanceID: "123",
+			Domain:     domainName,
+			Type:       domain.DomainTypeTrusted,
+			CreatedAt:  time.Now(),
+			UpdatedAt:  time.Now(),
+		})
+		require.NoError(t, err)
+
+		trustedDomainRemovedEvt := instance.NewTrustedDomainRemovedEvent(ctx, &instance.NewAggregate("123").Aggregate, domainName)
+
+		// Test
+		res := callReduce(t, ctx, rawTx, handler, trustedDomainRemovedEvt)
+		require.True(t, res)
+
+		// Verify
+		_, err = instanceDomainRepo.Get(ctx, tx,
+			database.WithCondition(
+				database.And(
+					instanceDomainRepo.InstanceIDCondition("123"),
+					instanceDomainRepo.DomainCondition(database.TextOperationEqual, domainName),
+					instanceDomainRepo.TypeCondition(domain.DomainTypeTrusted),
+				),
+			),
+		)
+		require.Error(t, err)
+		assert.ErrorIs(t, err, &database.NoRowFoundError{})
+	})
+}