refactor: invoker implementations and clean up code (#10962)

This pull request refactors the command execution and transaction
handling logic within the `domain` package. The primary goal is to
simplify the invoker chain, clarify transaction management, and improve
the overall structure for better maintainability and readability.

### Key Changes:

- **Simplified Invoker Chain**: The previous chain of invokers has been
refactored. A new `transactionInvoker` is introduced to explicitly
manage database transactions for commands that require them. This
responsibility is no longer spread across different parts of the
invocation flow.
- **`Transactional` Interface**: A new `Transactional` interface has
been introduced. Commands that need to run within a database transaction
now implement this interface. This provides a clear and explicit
contract for transactional operations.
- **Consolidated Transaction Logic**: The logic for starting,
committing, and rolling back transactions is now centralized within the
`transactionInvoker` and the `eventStoreInvoker`. The `EnsureTx` method
on `InvokeOpts` has been removed, and transaction handling is now
managed automatically by the invocation framework.
- **Improved Event Handling**: The `eventStoreInvoker` and its
`eventCollector` have been rewritten to be more robust and to correctly
handle nested and batched commands, ensuring that all events are
collected and published within the same transaction.
- **Code Cleanup**: A significant amount of now-unused code, including
the old `invoker_test.go`, has been removed. Various comments and
variable declarations have also been cleaned up.
- **Updated Command Implementations**: All commands in the `domain`
package (e.g., `ActivateOrgCommand`, `DeactivateOrgCommand`,
`DeleteOrgCommand`, `UpdateOrgCommand`) have been updated to use the new
`Transactional` interface where appropriate, removing manual transaction
management.
- **Enhanced Testing**: Tests have been added and updated to cover the
new transaction and event handling logic, ensuring the refactoring is
sound.
- **Use `PrimaryKeyCondition`**: Use the
`repository.PrimaryKeyCondition(...)` method instead of using
`database.And(...)` where possible.

These changes lead to a cleaner, more decoupled architecture for command
handling in ZITADEL.

---------

Co-authored-by: Marco A. <[email protected]>

backend/v3/domain/defaults.go

@@ -9,31 +9,17 @@ import (
 	"github.com/zitadel/zitadel/backend/v3/telemetry/tracing"
 )
 
-// The variables could also be moved to a struct.
-// I just started with the singleton pattern and kept it like this.
 var (
-	pool database.Pool
-	// userCodeAlgorithm crypto.EncryptionAlgorithm
+	pool             database.Pool
 	tracer           tracing.Tracer
 	logger           logging.Logger = *logging.NewLogger(slog.Default())
 	legacyEventstore eventstore.LegacyEventstore
-
-	// instanceCache cache.Cache[instanceCacheIndex, string, *Instance]
-	// orgCache cache.Cache[orgCacheIndex, string, *Org]
-
-	// generateID func() (string, error) = func() (string, error) {
-	// 	return strconv.FormatUint(rand.Uint64(), 10), nil
-	// }
 )
 
 func SetPool(p database.Pool) {
 	pool = p
 }
 
-// func SetUserCodeAlgorithm(algorithm crypto.EncryptionAlgorithm) {
-// 	userCodeAlgorithm = algorithm
-// }
-
 func SetTracer(t tracing.Tracer) {
 	tracer = t
 }
@@ -45,15 +31,3 @@ func SetLogger(l logging.Logger) {
 func SetLegacyEventstore(es eventstore.LegacyEventstore) {
 	legacyEventstore = es
 }
-
-// func SetUserRepository(repo func(database.QueryExecutor) UserRepository) {
-// 	userRepo = repo
-// }
-
-// func SetInstanceRepository(repo func(database.QueryExecutor) InstanceRepository) {
-// 	instanceRepo = repo
-// }
-
-// func SetCryptoRepository(repo func(database.QueryExecutor) CryptoRepository) {
-// 	cryptoRepo = repo
-// }

backend/v3/domain/executor_batch_test.go

@@ -1,9 +1,9 @@
 package domain_test
 
 import (
-	"context"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
 
@@ -12,34 +12,53 @@ import (
 )
 
 func TestBatchExecutors(t *testing.T) {
-	// Given
-	mockCtrl := gomock.NewController(t)
-	mockCmd1 := domainmock.NewMockCommander(mockCtrl)
-	mockCmd2 := domainmock.NewMockCommander(mockCtrl)
-	mockCmd3 := domainmock.NewMockCommander(mockCtrl)
+	t.Parallel()
+	t.Run("when commands are batched should run in expected order and return no error", func(t *testing.T) {
+		t.Parallel()
+		// Given
+		mockCtrl := gomock.NewController(t)
+		mockCmd1 := domainmock.NewMockCommander(mockCtrl)
+		mockCmd2 := domainmock.NewMockCommander(mockCtrl)
+		mockCmd3 := domainmock.NewMockCommander(mockCtrl)
 
-	mockCmd1.EXPECT().String().Return("cmd1").AnyTimes()
-	mockCmd2.EXPECT().String().Return("cmd2").AnyTimes()
-	mockCmd3.EXPECT().String().Return("cmd3").AnyTimes()
+		mockCmd1.EXPECT().String().Return("cmd1").AnyTimes()
+		mockCmd2.EXPECT().String().Return("cmd2").AnyTimes()
+		mockCmd3.EXPECT().String().Return("cmd3").AnyTimes()
 
-	gomock.InOrder(
-		mockCmd1.EXPECT().Validate(gomock.Any(), gomock.Any()).Times(1),
-		mockCmd1.EXPECT().Execute(gomock.Any(), gomock.Any()).Times(1),
-		mockCmd1.EXPECT().Events(gomock.Any(), gomock.Any()).Times(1),
-		mockCmd2.EXPECT().Validate(gomock.Any(), gomock.Any()).Times(1),
-		mockCmd2.EXPECT().Execute(gomock.Any(), gomock.Any()).Times(1),
-		mockCmd2.EXPECT().Events(gomock.Any(), gomock.Any()).Times(1),
-		mockCmd3.EXPECT().Validate(gomock.Any(), gomock.Any()).Times(1),
-		mockCmd3.EXPECT().Execute(gomock.Any(), gomock.Any()).Times(1),
-		mockCmd3.EXPECT().Events(gomock.Any(), gomock.Any()).Times(1),
-	)
+		gomock.InOrder(
+			mockCmd1.EXPECT().Validate(gomock.Any(), gomock.Any()).Times(1),
+			mockCmd1.EXPECT().Execute(gomock.Any(), gomock.Any()).Times(1),
+			mockCmd1.EXPECT().Events(gomock.Any(), gomock.Any()).Times(1),
+			mockCmd2.EXPECT().Validate(gomock.Any(), gomock.Any()).Times(1),
+			mockCmd2.EXPECT().Execute(gomock.Any(), gomock.Any()).Times(1),
+			mockCmd2.EXPECT().Events(gomock.Any(), gomock.Any()).Times(1),
+			mockCmd3.EXPECT().Validate(gomock.Any(), gomock.Any()).Times(1),
+			mockCmd3.EXPECT().Execute(gomock.Any(), gomock.Any()).Times(1),
+			mockCmd3.EXPECT().Events(gomock.Any(), gomock.Any()).Times(1),
+		)
 
-	batcher := domain.BatchExecutors(mockCmd1, mockCmd2, mockCmd3)
+		batcher := domain.BatchExecutors(mockCmd1, mockCmd2, mockCmd3)
+		err := domain.Invoke(t.Context(), batcher)
+		require.NoError(t, err)
+		require.True(t, mockCtrl.Satisfied())
+	})
 
-	// Test
-	err := domain.Invoke(context.Background(), batcher)
+	t.Run("when batched commands fail should return error from invoker", func(t *testing.T) {
+		t.Parallel()
+		mockCtrl := gomock.NewController(t)
+		mockCmd1 := domainmock.NewMockCommander(mockCtrl)
+		mockCmd2 := domainmock.NewMockCommander(mockCtrl)
 
-	// Verify
-	require.NoError(t, err)
-	require.True(t, mockCtrl.Satisfied())
+		mockCmd1.EXPECT().String().Return("cmd1").AnyTimes()
+		mockCmd2.EXPECT().String().Return("cmd2").AnyTimes()
+
+		gomock.InOrder(
+			mockCmd1.EXPECT().Validate(gomock.Any(), gomock.Any()).Times(1),
+			mockCmd1.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(assert.AnError).Times(1),
+		)
+
+		batcher := domain.BatchExecutors(mockCmd1, mockCmd2)
+		err := domain.Invoke(t.Context(), batcher)
+		require.ErrorIs(t, err, assert.AnError)
+	})
 }

backend/v3/domain/invoke.go

@@ -3,6 +3,8 @@ package domain
 import (
 	"context"
 	"fmt"
+
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
 )
 
 //go:generate mockgen -typed -package domainmock -destination ./mock/executor.mock.go . Executor
@@ -35,19 +37,20 @@ type Querier[T any] interface {
 
 // Invoke provides a way to execute commands within the domain package.
 // It uses a chain of responsibility pattern to handle the command execution.
-// The default chain includes logging, tracing, and event publishing.
-// If you want to invoke multiple commands in a single transaction, you can use the [batchExecutor].
+// The default chain includes logging, tracing, event publishing, transaction handling and validation.
+// If you want to invoke multiple commands in a single transaction, you can use the [BatchExecutors] helper.
 func Invoke(ctx context.Context, executor Executor, opts ...InvokeOpt) error {
-	invokeOpts := &InvokeOpts{
-		Invoker: newEventStoreInvoker(
-			newLoggingInvoker(
-				newTraceInvoker(
-					newValidatorInvoker(nil),
+	invokeOpts := DefaultOpts(
+		NewLoggingInvoker(
+			NewTraceInvoker(
+				NewEventStoreInvoker(
+					NewTransactionInvoker(
+						NewValidatorInvoker(nil),
+					),
 				),
 			),
 		),
-		DB: pool,
-	}
+	)
 	for _, opt := range opts {
 		opt(invokeOpts)
 	}
@@ -59,3 +62,39 @@ func Invoke(ctx context.Context, executor Executor, opts ...InvokeOpt) error {
 type Invoker interface {
 	Invoke(ctx context.Context, command Executor, opts *InvokeOpts) error
 }
+
+type invoker struct {
+	next Invoker
+}
+
+func (i invoker) execute(ctx context.Context, executor Executor, opts *InvokeOpts) error {
+	if i.next != nil {
+		return i.next.Invoke(ctx, executor, opts)
+	}
+	return executor.Execute(ctx, opts)
+}
+
+// ensureTx ensures that the InvokeOpts has a transaction.
+//
+// If [Commander] or [Querier] requires a transaction implement [Transactional]
+// the close function ends the transaction and resets [InvokeOpts].db.
+// If a new transaction is started, it returns a function to end the transaction.
+// The caller is responsible to call the returned function to end the transaction.
+// If no new transaction is started, the returned function is a no-op and still safe to call.
+func (i invoker) ensureTx(ctx context.Context, opts *InvokeOpts) (close func(err error) error, err error) {
+	beginner, ok := opts.DB().(database.Beginner)
+	if !ok {
+		return func(err error) error { return err }, nil
+	}
+
+	previousDB := opts.DB()
+	tx, err := beginner.Begin(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	opts.db = tx
+	return func(err error) error {
+		opts.db = previousDB
+		return tx.End(ctx, err)
+	}, nil
+}

backend/v3/domain/invoke_option.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/zitadel/zitadel/backend/v3/storage/database"
+	"github.com/zitadel/zitadel/backend/v3/storage/eventstore"
 )
 
 type InvokeOpt func(*InvokeOpts)
@@ -26,10 +27,33 @@ func WithProjectRepo(repo ProjectRepository) InvokeOpt {
 	}
 }
 
+// WithQueryExecutor sets the database client to be used by the command.
+// If not set, the default pool will be used.
+// This is mainly used for testing.
+func WithQueryExecutor(executor database.QueryExecutor) InvokeOpt {
+	return func(opts *InvokeOpts) {
+		opts.db = executor
+	}
+}
+
+// WithLegacyEventstore sets the eventstore to be used by the command.
+// If not set, the default one will be used.
+// This is mainly used for testing.
+func WithLegacyEventstore(es eventstore.LegacyEventstore) InvokeOpt {
+	return func(opts *InvokeOpts) {
+		opts.legacyEventstore = es
+	}
+}
+
 // InvokeOpts are passed to each command
-// they provide common fields used by commands like the database client.
 type InvokeOpts struct {
-	DB                     database.QueryExecutor
+	// db is the database client.
+	// [Executor]s MUST NOT access this field directly, use [InvokeOpts.DB] to access it.
+	//
+	// [Invoker]s may manipulate this field for example changing it to a transaction.
+	// Its their responsibility to restore it after ending the transaction.
+	db                     database.QueryExecutor
+	legacyEventstore       eventstore.LegacyEventstore
 	Invoker                Invoker
 	Permissions            PermissionChecker
 	organizationRepo       OrganizationRepository
@@ -37,61 +61,20 @@ type InvokeOpts struct {
 	projectRepo            ProjectRepository
 }
 
-type ensureTxOpts struct {
-	*database.TransactionOptions
+func (o *InvokeOpts) DB() database.QueryExecutor {
+	if o.db != nil {
+		return o.db
+	}
+	o.db = pool
+	return o.db
 }
 
-type EnsureTransactionOpt func(*ensureTxOpts)
-
-// EnsureTx ensures that the DB is a transaction. If it is not, it will start a new transaction.
-// The returned close function will end the transaction. If the DB is already a transaction, the close function
-// will be a noop because another [Executor] is already responsible for ending the transaction.
-func (o *InvokeOpts) EnsureTx(ctx context.Context, opts ...EnsureTransactionOpt) (close func(context.Context, error) error, err error) {
-	beginner, ok := o.DB.(database.Beginner)
-	if !ok {
-		// db is already a transaction
-		return func(_ context.Context, err error) error {
-			return err
-		}, nil
+func (o *InvokeOpts) LegacyEventstore() eventstore.LegacyEventstore {
+	if o.legacyEventstore != nil {
+		return o.legacyEventstore
 	}
-
-	txOpts := &ensureTxOpts{
-		TransactionOptions: new(database.TransactionOptions),
-	}
-	for _, opt := range opts {
-		opt(txOpts)
-	}
-
-	tx, err := beginner.Begin(ctx, txOpts.TransactionOptions)
-	if err != nil {
-		return nil, err
-	}
-	o.DB = tx
-
-	return func(ctx context.Context, err error) error {
-		return tx.End(ctx, err)
-	}, nil
-}
-
-// EnsureClient ensures that the o.DB is a client. If it is not, it will get a new client from the [database.Pool].
-// The returned close function will release the client. If the o.DB is already a client or transaction, the close function
-// will do nothing because another [Executor] is already responsible for releasing the client.
-func (o *InvokeOpts) EnsureClient(ctx context.Context) (close func(_ context.Context) error, err error) {
-	pool, ok := o.DB.(database.Pool)
-	if !ok {
-		// o.DB is already a client
-		return func(_ context.Context) error {
-			return nil
-		}, nil
-	}
-	client, err := pool.Acquire(ctx)
-	if err != nil {
-		return nil, err
-	}
-	o.DB = client
-	return func(ctx context.Context) error {
-		return client.Release(ctx)
-	}, nil
+	o.legacyEventstore = legacyEventstore
+	return o.legacyEventstore
 }
 
 func (o *InvokeOpts) Invoke(ctx context.Context, executor Executor) error {
@@ -106,7 +89,6 @@ func DefaultOpts(invoker Invoker) *InvokeOpts {
 		invoker = &noopInvoker{}
 	}
 	return &InvokeOpts{
-		DB:          pool,
 		Invoker:     invoker,
 		Permissions: &noopPermissionChecker{}, // prevent panics for now
 	}

backend/v3/domain/invoke_test.go

@@ -1,204 +0,0 @@
-package domain
-
-import (
-	"context"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"go.uber.org/mock/gomock"
-
-	"github.com/zitadel/zitadel/backend/v3/storage/database"
-	"github.com/zitadel/zitadel/backend/v3/storage/database/dbmock"
-	"github.com/zitadel/zitadel/backend/v3/storage/eventstore"
-	legacy_es "github.com/zitadel/zitadel/internal/eventstore"
-)
-
-type testLegacyEventstore struct{}
-
-func (es *testLegacyEventstore) PushWithNewClient(ctx context.Context, client database.QueryExecutor, commands ...legacy_es.Command) ([]legacy_es.Event, error) {
-	return nil, nil
-}
-
-var _ eventstore.LegacyEventstore = (*testLegacyEventstore)(nil)
-
-type invokeTestCommand struct {
-	events  []legacy_es.Command
-	execute func(ctx context.Context, opts *InvokeOpts) error
-}
-
-// Events implements [Commander].
-func (i *invokeTestCommand) Events(ctx context.Context, opts *InvokeOpts) ([]legacy_es.Command, error) {
-	return i.events, nil
-}
-
-// Execute implements [Commander].
-func (i *invokeTestCommand) Execute(ctx context.Context, opts *InvokeOpts) (err error) {
-	if i.execute == nil {
-		return nil
-	}
-	return i.execute(ctx, opts)
-}
-
-// String implements [Commander].
-func (i *invokeTestCommand) String() string {
-	return "invokeTestCommand"
-}
-
-// Validate implements [Commander].
-func (i *invokeTestCommand) Validate(ctx context.Context, opts *InvokeOpts) (err error) {
-	return nil
-}
-
-var _ Commander = (*invokeTestCommand)(nil)
-
-type invokeTestEvent struct {
-	id string
-}
-
-// Aggregate implements [legacy_es.Command].
-func (i *invokeTestEvent) Aggregate() *legacy_es.Aggregate {
-	return &legacy_es.Aggregate{
-		ID:            i.id,
-		Type:          "test",
-		ResourceOwner: "test",
-		InstanceID:    "test",
-		Version:       legacy_es.Version("v1"),
-	}
-}
-
-// Creator implements [legacy_es.Command].
-func (i *invokeTestEvent) Creator() string {
-	return "test"
-}
-
-// Fields implements [legacy_es.Command].
-func (i *invokeTestEvent) Fields() []*legacy_es.FieldOperation {
-	return nil
-}
-
-// Payload implements [legacy_es.Command].
-func (i *invokeTestEvent) Payload() any {
-	return nil
-}
-
-// Revision implements [legacy_es.Command].
-func (i *invokeTestEvent) Revision() uint16 {
-	return 1
-}
-
-// Type implements [legacy_es.Command].
-func (i *invokeTestEvent) Type() legacy_es.EventType {
-	return "test"
-}
-
-// UniqueConstraints implements [legacy_es.Command].
-func (i *invokeTestEvent) UniqueConstraints() []*legacy_es.UniqueConstraint {
-	return nil
-}
-
-var _ legacy_es.Command = (*invokeTestEvent)(nil)
-
-func Test_eventCollector_Invoke(t *testing.T) {
-	tests := []struct {
-		name                  string
-		command               Commander
-		expectedErr           error
-		assertCollectedEvents func(t *testing.T, events []legacy_es.Command)
-	}{
-		{
-			name:        "simple command with events",
-			expectedErr: nil,
-			command: &invokeTestCommand{
-				events: []legacy_es.Command{&invokeTestEvent{id: "1"}},
-			},
-			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
-				require.Len(t, events, 1)
-				assert.Equal(t, "1", events[0].Aggregate().ID)
-			},
-		},
-		{
-			name:        "simple command without events",
-			expectedErr: nil,
-			command:     &invokeTestCommand{},
-			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
-				assert.Len(t, events, 0)
-			},
-		},
-		{
-			name:        "command with sub commands",
-			expectedErr: nil,
-			command: &invokeTestCommand{
-				events: []legacy_es.Command{&invokeTestEvent{id: "1"}},
-				execute: func(ctx context.Context, opts *InvokeOpts) error {
-					return opts.Invoke(ctx, &invokeTestCommand{
-						events: []legacy_es.Command{&invokeTestEvent{id: "2"}},
-					})
-				},
-			},
-			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
-				require.Len(t, events, 2)
-				assert.Equal(t, "1", events[1].Aggregate().ID)
-				assert.Equal(t, "2", events[0].Aggregate().ID)
-			},
-		},
-		{
-			name:        "only sub commands with events",
-			expectedErr: nil,
-			command: &invokeTestCommand{
-				execute: func(ctx context.Context, opts *InvokeOpts) error {
-					return opts.Invoke(ctx, &invokeTestCommand{
-						events: []legacy_es.Command{&invokeTestEvent{id: "2"}},
-					})
-				},
-			},
-			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
-				require.Len(t, events, 1)
-				assert.Equal(t, "2", events[0].Aggregate().ID)
-			},
-		},
-		{
-			name:        "executor batch all commands",
-			expectedErr: nil,
-			command: &invokeTestCommand{
-				execute: func(ctx context.Context, opts *InvokeOpts) error {
-					return opts.Invoke(ctx, BatchExecutors(
-						&invokeTestCommand{
-							events: []legacy_es.Command{&invokeTestEvent{id: "2"}},
-						},
-						&invokeTestCommand{
-							events: []legacy_es.Command{&invokeTestEvent{id: "3"}},
-						},
-					))
-				},
-			},
-			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
-				require.Len(t, events, 2)
-				assert.Equal(t, "2", events[0].Aggregate().ID)
-				assert.Equal(t, "3", events[1].Aggregate().ID)
-			},
-		},
-	}
-
-	SetLegacyEventstore(new(testLegacyEventstore))
-
-	for _, tt := range tests {
-
-		t.Run(tt.name, func(t *testing.T) {
-			ctrl := gomock.NewController(t)
-			pool := dbmock.NewMockPool(ctrl)
-			tx := dbmock.NewMockTransaction(ctrl)
-			tx.EXPECT().End(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-			pool.EXPECT().Begin(gomock.Any(), gomock.Any()).Return(tx, nil).AnyTimes()
-
-			i := newEventStoreInvoker(nil)
-			opts := InvokeOpts{
-				Invoker: i,
-				DB:      pool,
-			}
-			err := opts.Invoke(t.Context(), tt.command)
-			require.ErrorIs(t, err, tt.expectedErr)
-			tt.assertCollectedEvents(t, i.collector.events)
-		})
-	}
-}

backend/v3/domain/invoker_event.go

@@ -10,12 +10,14 @@ import (
 // eventStoreInvoker checks if the [EventProducer].Events function returns any events.
 // If it does, it collects the events and publishes them to the event store.
 type eventStoreInvoker struct {
-	next      Invoker
-	collector *eventCollector
+	invoker
+	collector eventCollector
 }
 
-func newEventStoreInvoker(next Invoker) *eventStoreInvoker {
-	return &eventStoreInvoker{next: next}
+func NewEventStoreInvoker(next Invoker) *eventStoreInvoker {
+	return &eventStoreInvoker{
+		invoker: invoker{next: next},
+	}
 }
 
 type EventProducer interface {
@@ -25,82 +27,140 @@ type EventProducer interface {
 }
 
 func (i *eventStoreInvoker) Invoke(ctx context.Context, executor Executor, opts *InvokeOpts) (err error) {
+	if _, ok := executor.(*batchExecutor); ok {
+		return i.collect(ctx, executor, opts)
+	}
 	if _, ok := executor.(EventProducer); !ok {
 		return i.execute(ctx, executor, opts)
 	}
+	return i.collect(ctx, executor, opts)
+}
 
-	if i.collector != nil {
-		return i.collector.Invoke(ctx, executor, opts)
-	}
-
-	close, err := opts.EnsureTx(ctx)
-	if err != nil {
-		return err
-	}
-	defer func() { err = close(ctx, err) }()
-
-	i.collector = &eventCollector{next: i.next}
-
-	if err = i.collector.Invoke(ctx, executor, opts); err != nil {
-		return err
-	}
-
-	if len(i.collector.events) > 0 {
-		err = eventstore.Publish(ctx, legacyEventstore, opts.DB, i.collector.events...)
+func (i *eventStoreInvoker) collect(ctx context.Context, executor Executor, opts *InvokeOpts) (err error) {
+	if i.collector == nil {
+		var close func(err error) error
+		close, err = i.ensureTx(ctx, opts)
 		if err != nil {
 			return err
 		}
+		defer func() {
+			if err != nil {
+				err = close(err)
+				return
+			}
+			commands := i.collector.commands()
+			if len(commands) == 0 {
+				err = close(nil)
+				return
+			}
+			err = eventstore.Publish(ctx, opts.LegacyEventstore(), opts.DB(), commands...)
+			err = close(err)
+		}()
+		i.collector = initEventCollector(i.invoker, nil, executor)
+	} else {
+		i.collector = i.collector.initSub(executor)
 	}
+
+	defer func() {
+		if i.collector.parent() == nil {
+			return
+		}
+		i.collector = i.collector.parent()
+	}()
+	err = i.execute(ctx, executor, opts)
+	if err != nil {
+		return err
+	}
+	return i.collector.collect(ctx, opts)
+}
+
+type eventCollector interface {
+	collect(ctx context.Context, opts *InvokeOpts) error
+	commands() []legacy_es.Command
+	initSub(executor Executor) eventCollector
+	parent() eventCollector
+}
+
+type batchCollector struct {
+	invoker
+	executor   Executor
+	collectors []eventCollector
+	previous   eventCollector
+}
+
+// collect implements [eventCollector].
+func (b batchCollector) collect(ctx context.Context, opts *InvokeOpts) error {
 	return nil
 }
 
-func (i *eventStoreInvoker) execute(ctx context.Context, executor Executor, opts *InvokeOpts) error {
-	if i.next != nil {
-		return i.next.Invoke(ctx, executor, opts)
+// commands implements [eventCollector].
+func (b batchCollector) commands() []legacy_es.Command {
+	var commands []legacy_es.Command
+	for _, collector := range b.collectors {
+		commands = append(commands, collector.commands()...)
 	}
-	return executor.Execute(ctx, opts)
+	return commands
 }
 
-// eventCollector collects events from all commands. The [eventStoreInvoker] pushes the collected events after all commands are executed.
-// The events are collected after the command got executed, the collector ensures that the command is executed in the same transaction as writing the events.
-type eventCollector struct {
-	next          Invoker
-	events        []legacy_es.Command
-	shouldPrepend bool
+// initSub implements [eventCollector].
+func (b *batchCollector) initSub(executor Executor) eventCollector {
+	b.collectors = append(b.collectors, initEventCollector(b.invoker, b, executor))
+	return b.collectors[len(b.collectors)-1]
 }
 
-func (i *eventCollector) Invoke(ctx context.Context, executor Executor, opts *InvokeOpts) (err error) {
-	command, ok := executor.(EventProducer)
-	if !ok {
-		if i.next != nil {
-			return i.next.Invoke(ctx, executor, opts)
-		}
-		return executor.Execute(ctx, opts)
-	}
+// parent implements [eventCollector].
+func (b batchCollector) parent() eventCollector {
+	return b.previous
+}
 
-	shouldPrepend := i.shouldPrepend
-	i.shouldPrepend = false
+var _ eventCollector = (*batchCollector)(nil)
 
-	if i.next != nil {
-		i.shouldPrepend = true
-		err = i.next.Invoke(ctx, executor, opts)
-		i.shouldPrepend = false
-	} else {
-		err = executor.Execute(ctx, opts)
-	}
-	if err != nil {
-		return err
-	}
-	collectedEvents, err := command.Events(ctx, opts)
-	if err != nil {
-		return err
-	}
-
-	if shouldPrepend {
-		i.events = append(collectedEvents, i.events...)
-	} else {
-		i.events = append(i.events, collectedEvents...)
-	}
+type commandCollector struct {
+	invoker
+	producer EventProducer
+	sub      eventCollector
+	previous eventCollector
+	cmds     []legacy_es.Command
+}
 
+// collect implements [eventCollector].
+func (c *commandCollector) collect(ctx context.Context, opts *InvokeOpts) (err error) {
+	c.cmds, err = c.producer.Events(ctx, opts)
 	return err
 }
+
+// commands implements [eventCollector].
+func (c commandCollector) commands() []legacy_es.Command {
+	if c.sub == nil {
+		return c.cmds
+	}
+	return append(c.cmds, c.sub.commands()...)
+}
+
+// initSub implements [eventCollector].
+func (c *commandCollector) initSub(executor Executor) eventCollector {
+	c.sub = initEventCollector(c.invoker, c, executor)
+	return c.sub
+}
+
+// parent implements [eventCollector].
+func (c commandCollector) parent() eventCollector {
+	return c.previous
+}
+
+var _ eventCollector = (*commandCollector)(nil)
+
+func initEventCollector(i invoker, parent eventCollector, executor Executor) eventCollector {
+	if _, ok := executor.(*batchExecutor); ok {
+		return &batchCollector{
+			invoker:  i,
+			executor: executor,
+			previous: parent,
+		}
+	}
+	return &commandCollector{
+		invoker:  i,
+		producer: executor.(EventProducer),
+		previous: parent,
+	}
+}

backend/v3/domain/invoker_event_test.go

@@ -0,0 +1,376 @@
+package domain_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+
+	"github.com/zitadel/zitadel/backend/v3/domain"
+	domainmock "github.com/zitadel/zitadel/backend/v3/domain/mock"
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+	"github.com/zitadel/zitadel/backend/v3/storage/database/dbmock"
+	"github.com/zitadel/zitadel/backend/v3/storage/eventstore"
+	legacy_es "github.com/zitadel/zitadel/internal/eventstore"
+)
+
+type testLegacyEventstore struct {
+	commands []legacy_es.Command
+}
+
+func (es *testLegacyEventstore) PushWithNewClient(ctx context.Context, client database.QueryExecutor, commands ...legacy_es.Command) ([]legacy_es.Event, error) {
+	es.commands = append(es.commands, commands...)
+	return nil, nil
+}
+
+var _ eventstore.LegacyEventstore = (*testLegacyEventstore)(nil)
+
+type invokeTestEvent struct {
+	id string
+}
+
+// Aggregate implements [legacy_es.Command].
+func (i *invokeTestEvent) Aggregate() *legacy_es.Aggregate {
+	return &legacy_es.Aggregate{
+		ID:            i.id,
+		Type:          "test",
+		ResourceOwner: "test",
+		InstanceID:    "test",
+		Version:       legacy_es.Version("v1"),
+	}
+}
+
+// Creator implements [legacy_es.Command].
+func (i *invokeTestEvent) Creator() string {
+	return "test"
+}
+
+// Fields implements [legacy_es.Command].
+func (i *invokeTestEvent) Fields() []*legacy_es.FieldOperation {
+	return nil
+}
+
+// Payload implements [legacy_es.Command].
+func (i *invokeTestEvent) Payload() any {
+	return nil
+}
+
+// Revision implements [legacy_es.Command].
+func (i *invokeTestEvent) Revision() uint16 {
+	return 1
+}
+
+// Type implements [legacy_es.Command].
+func (i *invokeTestEvent) Type() legacy_es.EventType {
+	return "test"
+}
+
+// UniqueConstraints implements [legacy_es.Command].
+func (i *invokeTestEvent) UniqueConstraints() []*legacy_es.UniqueConstraint {
+	return nil
+}
+
+var _ legacy_es.Command = (*invokeTestEvent)(nil)
+
+func Test_eventCollector_Invoke(t *testing.T) {
+	tests := []struct {
+		name                  string
+		command               func(ctrl *gomock.Controller) domain.Executor
+		queryExecutor         func(ctrl *gomock.Controller) database.QueryExecutor
+		expectedErr           error
+		assertCollectedEvents func(t *testing.T, events []legacy_es.Command)
+	}{
+		{
+			name:        "simple command without events",
+			expectedErr: nil,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				cmd := domainmock.NewMockCommander(ctrl)
+				gomock.InOrder(
+					cmd.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					cmd.EXPECT().Events(gomock.Any(), gomock.Any()).Return(nil, nil),
+				)
+				return cmd
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				assert.Len(t, events, 0)
+			},
+		},
+		{
+			name:        "simple command with events",
+			expectedErr: nil,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				cmd := domainmock.NewMockCommander(ctrl)
+				gomock.InOrder(
+					cmd.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					cmd.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "1"}}, nil),
+				)
+				return cmd
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				require.Len(t, events, 1)
+				assert.Equal(t, "1", events[0].Aggregate().ID)
+			},
+		},
+		{
+			name:        "command with sub commands",
+			expectedErr: nil,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				mainCmd := domainmock.NewMockCommander(ctrl)
+				subCmd := domainmock.NewMockCommander(ctrl)
+				gomock.InOrder(
+					mainCmd.EXPECT().Execute(gomock.Any(), gomock.Any()).DoAndReturn(
+						func(ctx context.Context, opts *domain.InvokeOpts) error {
+							return opts.Invoke(ctx, subCmd)
+						},
+					),
+					subCmd.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					subCmd.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "2"}}, nil),
+					mainCmd.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "1"}}, nil),
+				)
+				return mainCmd
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				require.Len(t, events, 2)
+				assert.Equal(t, "1", events[0].Aggregate().ID)
+				assert.Equal(t, "2", events[1].Aggregate().ID)
+			},
+		},
+		{
+			name:        "command with multiple sub commands",
+			expectedErr: nil,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				mainCmd := domainmock.NewMockCommander(ctrl)
+				firstSubCmd := domainmock.NewMockCommander(ctrl)
+				secondSubCommand := domainmock.NewMockCommander(ctrl)
+				gomock.InOrder(
+					mainCmd.EXPECT().Execute(gomock.Any(), gomock.Any()).DoAndReturn(
+						func(ctx context.Context, opts *domain.InvokeOpts) error {
+							return opts.Invoke(ctx, firstSubCmd)
+						},
+					),
+					firstSubCmd.EXPECT().Execute(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, io *domain.InvokeOpts) error {
+						return io.Invoke(ctx, secondSubCommand)
+					}),
+					secondSubCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					secondSubCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "3"}}, nil),
+					firstSubCmd.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "2"}}, nil),
+					mainCmd.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "1"}}, nil),
+				)
+				return mainCmd
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				require.Len(t, events, 3)
+				assert.Equal(t, "1", events[0].Aggregate().ID)
+				assert.Equal(t, "2", events[1].Aggregate().ID)
+				assert.Equal(t, "3", events[2].Aggregate().ID)
+			},
+		},
+		{
+			name:        "only sub commands with events",
+			expectedErr: nil,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				mainCmd := domainmock.NewMockCommander(ctrl)
+				subCmd := domainmock.NewMockCommander(ctrl)
+				gomock.InOrder(
+					mainCmd.EXPECT().Execute(gomock.Any(), gomock.Any()).DoAndReturn(
+						func(ctx context.Context, opts *domain.InvokeOpts) error {
+							return opts.Invoke(ctx, subCmd)
+						},
+					),
+					subCmd.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					subCmd.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "2"}}, nil),
+					mainCmd.EXPECT().Events(gomock.Any(), gomock.Any()).Return(nil, nil),
+				)
+				return mainCmd
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				require.Len(t, events, 1)
+				assert.Equal(t, "2", events[0].Aggregate().ID)
+			},
+		},
+		{
+			name:        "executor batch all commands",
+			expectedErr: nil,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				mainCommand := domainmock.NewMockCommander(ctrl)
+				firstSubCommand := domainmock.NewMockCommander(ctrl)
+				secondSubCommand := domainmock.NewMockCommander(ctrl)
+
+				gomock.InOrder(
+					mainCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).DoAndReturn(
+						func(ctx context.Context, opts *domain.InvokeOpts) error {
+							return opts.Invoke(ctx,
+								domain.BatchExecutors(
+									firstSubCommand,
+									secondSubCommand,
+								),
+							)
+						},
+					),
+					firstSubCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					firstSubCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "2"}}, nil),
+					secondSubCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					secondSubCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "3"}}, nil),
+					mainCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "1"}}, nil),
+				)
+
+				return mainCommand
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				require.Len(t, events, 3)
+				assert.Equal(t, "1", events[0].Aggregate().ID)
+				assert.Equal(t, "2", events[1].Aggregate().ID)
+				assert.Equal(t, "3", events[2].Aggregate().ID)
+			},
+		},
+		{
+			name:        "executor batch with sub batch",
+			expectedErr: nil,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				mainCommand := domainmock.NewMockCommander(ctrl)
+				firstSubCommand := domainmock.NewMockCommander(ctrl)
+				secondSubCommand := domainmock.NewMockCommander(ctrl)
+				thirdSubCommand := domainmock.NewMockCommander(ctrl)
+				fourthSubCommand := domainmock.NewMockCommander(ctrl)
+				fifthSubCommand := domainmock.NewMockCommander(ctrl)
+
+				gomock.InOrder(
+					mainCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).DoAndReturn(
+						func(ctx context.Context, opts *domain.InvokeOpts) error {
+							return opts.Invoke(ctx,
+								domain.BatchExecutors(
+									firstSubCommand,
+									domain.BatchExecutors(
+										secondSubCommand,
+										thirdSubCommand,
+									),
+								),
+							)
+						},
+					),
+					firstSubCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					firstSubCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "2"}}, nil),
+					secondSubCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					secondSubCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "3"}}, nil),
+					thirdSubCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).DoAndReturn(
+						func(ctx context.Context, opts *domain.InvokeOpts) error {
+							return opts.Invoke(ctx,
+								domain.BatchExecutors(
+									domain.BatchExecutors(
+										fourthSubCommand,
+									),
+									domain.BatchExecutors(
+										fifthSubCommand,
+									),
+								),
+							)
+						},
+					),
+					fourthSubCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					fourthSubCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "5"}}, nil),
+					fifthSubCommand.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					fifthSubCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "6"}}, nil),
+					thirdSubCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "4"}}, nil),
+					mainCommand.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "1"}}, nil),
+				)
+				return mainCommand
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				require.Len(t, events, 6)
+				assert.Equal(t, "1", events[0].Aggregate().ID)
+				assert.Equal(t, "2", events[1].Aggregate().ID)
+				assert.Equal(t, "3", events[2].Aggregate().ID)
+				assert.Equal(t, "4", events[3].Aggregate().ID)
+				assert.Equal(t, "5", events[4].Aggregate().ID)
+				assert.Equal(t, "6", events[5].Aggregate().ID)
+			},
+		},
+		{
+			name:        "command with sub query",
+			expectedErr: nil,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				mainCmd := domainmock.NewMockCommander(ctrl)
+				subQuery := domainmock.NewMockQuerier[string](ctrl)
+				gomock.InOrder(
+					mainCmd.EXPECT().Execute(gomock.Any(), gomock.Any()).DoAndReturn(
+						func(ctx context.Context, opts *domain.InvokeOpts) error {
+							return opts.Invoke(ctx, subQuery)
+						},
+					),
+					subQuery.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil),
+					mainCmd.EXPECT().Events(gomock.Any(), gomock.Any()).Return([]legacy_es.Command{&invokeTestEvent{id: "1"}}, nil),
+				)
+				return mainCmd
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				require.Len(t, events, 1)
+				assert.Equal(t, "1", events[0].Aggregate().ID)
+			},
+		},
+		{
+			name:        "begin fails",
+			expectedErr: assert.AnError,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				return domainmock.NewMockCommander(ctrl)
+			},
+			queryExecutor: func(ctrl *gomock.Controller) database.QueryExecutor {
+				pool := dbmock.NewMockPool(ctrl)
+				pool.EXPECT().Begin(gomock.Any(), gomock.Any()).Return(nil, assert.AnError).Times(1)
+				return pool
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				require.Len(t, events, 0)
+			},
+		},
+		{
+			name:        "execute fails",
+			expectedErr: assert.AnError,
+			command: func(ctrl *gomock.Controller) domain.Executor {
+				cmd := domainmock.NewMockCommander(ctrl)
+				cmd.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(assert.AnError)
+				return cmd
+			},
+			queryExecutor: func(ctrl *gomock.Controller) database.QueryExecutor {
+				pool := dbmock.NewMockPool(ctrl)
+				tx := dbmock.NewMockTransaction(ctrl)
+				gomock.InOrder(
+					pool.EXPECT().Begin(gomock.Any(), gomock.Any()).Return(tx, nil).Times(1),
+					tx.EXPECT().End(gomock.Any(), assert.AnError).Return(assert.AnError).Times(1),
+				)
+				return pool
+			},
+			assertCollectedEvents: func(t *testing.T, events []legacy_es.Command) {
+				require.Len(t, events, 0)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+
+			i := domain.NewEventStoreInvoker(nil)
+			opts := domain.InvokeOpts{
+				Invoker: i,
+			}
+			if tt.queryExecutor != nil {
+				domain.WithQueryExecutor(tt.queryExecutor(ctrl))(&opts)
+			} else {
+				pool := dbmock.NewMockPool(ctrl)
+				tx := dbmock.NewMockTransaction(ctrl)
+				tx.EXPECT().End(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+				pool.EXPECT().Begin(gomock.Any(), gomock.Any()).Return(tx, nil).Times(1)
+				domain.WithQueryExecutor(pool)(&opts)
+			}
+
+			eventStore := new(testLegacyEventstore)
+			domain.WithLegacyEventstore(eventStore)(&opts)
+
+			err := opts.Invoke(t.Context(), tt.command(ctrl))
+			require.ErrorIs(t, err, tt.expectedErr)
+			tt.assertCollectedEvents(t, eventStore.commands)
+		})
+	}
+}

backend/v3/domain/invoker_logging.go

@@ -8,23 +8,20 @@ import (
 // loggingInvoker decorates each command with logging.
 // It is an example implementation and logs the command name at the beginning and success or failure after the command got executed.
 type loggingInvoker struct {
-	next Invoker
+	invoker
 }
 
-func newLoggingInvoker(next Invoker) *loggingInvoker {
-	return &loggingInvoker{next: next}
+func NewLoggingInvoker(next Invoker) *loggingInvoker {
+	return &loggingInvoker{
+		invoker: invoker{next: next},
+	}
 }
 
 func (i *loggingInvoker) Invoke(ctx context.Context, executor Executor, opts *InvokeOpts) (err error) {
 	start := time.Now()
 	logger.InfoContext(ctx, "invoking", "name", executor.String())
 
-	if i.next != nil {
-		err = i.next.Invoke(ctx, executor, opts)
-	} else {
-		err = executor.Execute(ctx, opts)
-	}
-
+	err = i.execute(ctx, executor, opts)
 	if err != nil {
 		logger.ErrorContext(ctx, "invocation failed", "name", executor.String(), "error", err, "took", time.Since(start))
 		return err

backend/v3/domain/invoker_noop.go

@@ -3,12 +3,9 @@ package domain
 import "context"
 
 type noopInvoker struct {
-	next Invoker
+	invoker
 }
 
-func (i *noopInvoker) Invoke(ctx context.Context, command Executor, opts *InvokeOpts) error {
-	if i.next != nil {
-		return i.next.Invoke(ctx, command, opts)
-	}
-	return command.Execute(ctx, opts)
+func (i *noopInvoker) Invoke(ctx context.Context, executor Executor, opts *InvokeOpts) error {
+	return i.execute(ctx, executor, opts)
 }

backend/v3/domain/invoker_tracing.go

@@ -7,11 +7,13 @@ import (
 
 // traceInvoker decorates each command with tracing.
 type traceInvoker struct {
-	next Invoker
+	invoker
 }
 
-func newTraceInvoker(next Invoker) *traceInvoker {
-	return &traceInvoker{next: next}
+func NewTraceInvoker(next Invoker) *traceInvoker {
+	return &traceInvoker{
+		invoker: invoker{next: next},
+	}
 }
 
 func (i *traceInvoker) Invoke(ctx context.Context, executor Executor, opts *InvokeOpts) (err error) {
@@ -23,8 +25,5 @@ func (i *traceInvoker) Invoke(ctx context.Context, executor Executor, opts *Invo
 		span.End()
 	}()
 
-	if i.next != nil {
-		return i.next.Invoke(ctx, executor, opts)
-	}
-	return executor.Execute(ctx, opts)
+	return i.execute(ctx, executor, opts)
 }

backend/v3/domain/invoker_transaction.go

@@ -0,0 +1,40 @@
+package domain
+
+import (
+	"context"
+)
+
+// transactionInvoker ensures that [InvokeOpts].DB() returns a [database.Transaction].
+// if a new transaction is started, it will be committed or rolled back after the [Executor] execution.
+type transactionInvoker struct {
+	invoker
+}
+
+// NewTransactionInvoker creates a new [transactionInvoker].
+func NewTransactionInvoker(next Invoker) *transactionInvoker {
+	return &transactionInvoker{
+		invoker: invoker{next: next},
+	}
+}
+
+// Transactional is implemented by [Commander]s and [Querier]s wishing to access the DB inside a transaction.
+//
+//go:generate mockgen -destination=mock/transactional.mock.go -package=domainmock . Transactional
+type Transactional interface {
+	RequiresTransaction()
+}
+
+// Invoke implements [Invoker].
+func (i *transactionInvoker) Invoke(ctx context.Context, executor Executor, opts *InvokeOpts) (err error) {
+	if _, ok := executor.(Transactional); !ok {
+		return i.execute(ctx, executor, opts)
+	}
+
+	close, err := i.ensureTx(ctx, opts)
+	if err != nil {
+		return err
+	}
+	defer func() { err = close(err) }()
+
+	return i.execute(ctx, executor, opts)
+}

backend/v3/domain/invoker_transaction_test.go

@@ -0,0 +1,95 @@
+package domain_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+
+	"github.com/zitadel/zitadel/backend/v3/domain"
+	domainmock "github.com/zitadel/zitadel/backend/v3/domain/mock"
+	"github.com/zitadel/zitadel/backend/v3/storage/database"
+	"github.com/zitadel/zitadel/backend/v3/storage/database/dbmock"
+)
+
+type testTransactionalExecutor struct {
+	domain.Transactional
+	domain.Executor
+}
+
+func Test_transactionInvoker_Invoke(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name     string
+		executor func(ctrl *gomock.Controller) domain.Executor
+		db       func(ctrl *gomock.Controller) database.Pool
+		wantErr  error
+	}{
+		{
+			name: "non-transactional executor does not start a transaction",
+			executor: func(ctrl *gomock.Controller) domain.Executor {
+				mock := domainmock.NewMockExecutor(ctrl)
+				mock.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil)
+				return mock
+			},
+			db: func(ctrl *gomock.Controller) database.Pool {
+				return dbmock.NewMockPool(ctrl)
+			},
+			wantErr: nil,
+		},
+		{
+			name: "transactional executor starts a transaction execution successful",
+			executor: func(ctrl *gomock.Controller) domain.Executor {
+				transactional := domainmock.NewMockTransactional(ctrl)
+				executor := domainmock.NewMockExecutor(ctrl)
+				executor.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(nil)
+				return &testTransactionalExecutor{
+					Transactional: transactional,
+					Executor:      executor,
+				}
+			},
+			db: func(ctrl *gomock.Controller) database.Pool {
+				pool := dbmock.NewMockPool(ctrl)
+				tx := dbmock.NewMockTransaction(ctrl)
+				pool.EXPECT().Begin(gomock.Any(), gomock.Any()).Return(tx, nil)
+				tx.EXPECT().End(gomock.Any(), nil).Return(nil)
+				return pool
+			},
+			wantErr: nil,
+		},
+		{
+			name: "transactional executor starts a transaction execution failed",
+			executor: func(ctrl *gomock.Controller) domain.Executor {
+				transactional := domainmock.NewMockTransactional(ctrl)
+				executor := domainmock.NewMockExecutor(ctrl)
+				executor.EXPECT().Execute(gomock.Any(), gomock.Any()).Return(assert.AnError)
+				return &testTransactionalExecutor{
+					Transactional: transactional,
+					Executor:      executor,
+				}
+			},
+			db: func(ctrl *gomock.Controller) database.Pool {
+				pool := dbmock.NewMockPool(ctrl)
+				tx := dbmock.NewMockTransaction(ctrl)
+				pool.EXPECT().Begin(gomock.Any(), gomock.Any()).Return(tx, nil)
+				tx.EXPECT().End(gomock.Any(), assert.AnError).DoAndReturn(func(ctx context.Context, err error) error { return err })
+				return pool
+			},
+			wantErr: assert.AnError,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			ctrl := gomock.NewController(t)
+			opts := &domain.InvokeOpts{}
+			domain.WithQueryExecutor(tt.db(ctrl))(opts)
+
+			invoker := domain.NewTransactionInvoker(nil)
+			gotErr := invoker.Invoke(t.Context(), tt.executor(ctrl), opts)
+			require.ErrorIs(t, gotErr, tt.wantErr)
+		})
+	}
+}

backend/v3/domain/invoker_validation.go

@@ -5,11 +5,15 @@ import (
 )
 
 type validatorInvoker struct {
-	next Invoker
+	invoker
 }
 
-func newValidatorInvoker(next Invoker) *validatorInvoker {
-	return &validatorInvoker{next: next}
+// NewValidatorInvoker creates a new [validatorInvoker].
+// It is responsible for calling the [Validator.Validate] function before executing the [Executor].
+func NewValidatorInvoker(next Invoker) *validatorInvoker {
+	return &validatorInvoker{
+		invoker: invoker{next: next},
+	}
 }
 
 type Validator interface {
@@ -23,9 +27,5 @@ func (i *validatorInvoker) Invoke(ctx context.Context, executor Executor, opts *
 		}
 	}
 
-	if i.next != nil {
-		return i.next.Invoke(ctx, executor, opts)
-	}
-
-	return executor.Execute(ctx, opts)
+	return i.execute(ctx, executor, opts)
 }

backend/v3/domain/mock/executor.mock.go

@@ -0,0 +1,118 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: github.com/zitadel/zitadel/backend/v3/domain (interfaces: Executor)
+//
+// Generated by this command:
+//
+//	mockgen -typed -package domainmock -destination ./mock/executor.mock.go . Executor
+//
+
+// Package domainmock is a generated GoMock package.
+package domainmock
+
+import (
+	context "context"
+	reflect "reflect"
+
+	domain "github.com/zitadel/zitadel/backend/v3/domain"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockExecutor is a mock of Executor interface.
+type MockExecutor struct {
+	ctrl     *gomock.Controller
+	recorder *MockExecutorMockRecorder
+	isgomock struct{}
+}
+
+// MockExecutorMockRecorder is the mock recorder for MockExecutor.
+type MockExecutorMockRecorder struct {
+	mock *MockExecutor
+}
+
+// NewMockExecutor creates a new mock instance.
+func NewMockExecutor(ctrl *gomock.Controller) *MockExecutor {
+	mock := &MockExecutor{ctrl: ctrl}
+	mock.recorder = &MockExecutorMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockExecutor) EXPECT() *MockExecutorMockRecorder {
+	return m.recorder
+}
+
+// Execute mocks base method.
+func (m *MockExecutor) Execute(ctx context.Context, opts *domain.InvokeOpts) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Execute", ctx, opts)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Execute indicates an expected call of Execute.
+func (mr *MockExecutorMockRecorder) Execute(ctx, opts any) *MockExecutorExecuteCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Execute", reflect.TypeOf((*MockExecutor)(nil).Execute), ctx, opts)
+	return &MockExecutorExecuteCall{Call: call}
+}
+
+// MockExecutorExecuteCall wrap *gomock.Call
+type MockExecutorExecuteCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockExecutorExecuteCall) Return(err error) *MockExecutorExecuteCall {
+	c.Call = c.Call.Return(err)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockExecutorExecuteCall) Do(f func(context.Context, *domain.InvokeOpts) error) *MockExecutorExecuteCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockExecutorExecuteCall) DoAndReturn(f func(context.Context, *domain.InvokeOpts) error) *MockExecutorExecuteCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// String mocks base method.
+func (m *MockExecutor) String() string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "String")
+	ret0, _ := ret[0].(string)
+	return ret0
+}
+
+// String indicates an expected call of String.
+func (mr *MockExecutorMockRecorder) String() *MockExecutorStringCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "String", reflect.TypeOf((*MockExecutor)(nil).String))
+	return &MockExecutorStringCall{Call: call}
+}
+
+// MockExecutorStringCall wrap *gomock.Call
+type MockExecutorStringCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockExecutorStringCall) Return(arg0 string) *MockExecutorStringCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockExecutorStringCall) Do(f func() string) *MockExecutorStringCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockExecutorStringCall) DoAndReturn(f func() string) *MockExecutorStringCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}

backend/v3/domain/mock/org.mock.go

@@ -316,17 +316,17 @@ func (c *MockOrganizationRepositoryIDColumnCall) DoAndReturn(f func() database.C
 }
 
 // IDCondition mocks base method.
-func (m *MockOrganizationRepository) IDCondition(instanceID string) database.Condition {
+func (m *MockOrganizationRepository) IDCondition(orgID string) database.Condition {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "IDCondition", instanceID)
+	ret := m.ctrl.Call(m, "IDCondition", orgID)
 	ret0, _ := ret[0].(database.Condition)
 	return ret0
 }
 
 // IDCondition indicates an expected call of IDCondition.
-func (mr *MockOrganizationRepositoryMockRecorder) IDCondition(instanceID any) *MockOrganizationRepositoryIDConditionCall {
+func (mr *MockOrganizationRepositoryMockRecorder) IDCondition(orgID any) *MockOrganizationRepositoryIDConditionCall {
 	mr.mock.ctrl.T.Helper()
-	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IDCondition", reflect.TypeOf((*MockOrganizationRepository)(nil).IDCondition), instanceID)
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IDCondition", reflect.TypeOf((*MockOrganizationRepository)(nil).IDCondition), orgID)
 	return &MockOrganizationRepositoryIDConditionCall{Call: call}
 }
 
@@ -625,6 +625,82 @@ func (c *MockOrganizationRepositoryNameConditionCall) DoAndReturn(f func(databas
 	return c
 }
 
+// PrimaryKeyColumns mocks base method.
+func (m *MockOrganizationRepository) PrimaryKeyColumns() []database.Column {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "PrimaryKeyColumns")
+	ret0, _ := ret[0].([]database.Column)
+	return ret0
+}
+
+// PrimaryKeyColumns indicates an expected call of PrimaryKeyColumns.
+func (mr *MockOrganizationRepositoryMockRecorder) PrimaryKeyColumns() *MockOrganizationRepositoryPrimaryKeyColumnsCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PrimaryKeyColumns", reflect.TypeOf((*MockOrganizationRepository)(nil).PrimaryKeyColumns))
+	return &MockOrganizationRepositoryPrimaryKeyColumnsCall{Call: call}
+}
+
+// MockOrganizationRepositoryPrimaryKeyColumnsCall wrap *gomock.Call
+type MockOrganizationRepositoryPrimaryKeyColumnsCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockOrganizationRepositoryPrimaryKeyColumnsCall) Return(arg0 []database.Column) *MockOrganizationRepositoryPrimaryKeyColumnsCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockOrganizationRepositoryPrimaryKeyColumnsCall) Do(f func() []database.Column) *MockOrganizationRepositoryPrimaryKeyColumnsCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockOrganizationRepositoryPrimaryKeyColumnsCall) DoAndReturn(f func() []database.Column) *MockOrganizationRepositoryPrimaryKeyColumnsCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// PrimaryKeyCondition mocks base method.
+func (m *MockOrganizationRepository) PrimaryKeyCondition(instanceID, organizationID string) database.Condition {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "PrimaryKeyCondition", instanceID, organizationID)
+	ret0, _ := ret[0].(database.Condition)
+	return ret0
+}
+
+// PrimaryKeyCondition indicates an expected call of PrimaryKeyCondition.
+func (mr *MockOrganizationRepositoryMockRecorder) PrimaryKeyCondition(instanceID, organizationID any) *MockOrganizationRepositoryPrimaryKeyConditionCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PrimaryKeyCondition", reflect.TypeOf((*MockOrganizationRepository)(nil).PrimaryKeyCondition), instanceID, organizationID)
+	return &MockOrganizationRepositoryPrimaryKeyConditionCall{Call: call}
+}
+
+// MockOrganizationRepositoryPrimaryKeyConditionCall wrap *gomock.Call
+type MockOrganizationRepositoryPrimaryKeyConditionCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockOrganizationRepositoryPrimaryKeyConditionCall) Return(arg0 database.Condition) *MockOrganizationRepositoryPrimaryKeyConditionCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockOrganizationRepositoryPrimaryKeyConditionCall) Do(f func(string, string) database.Condition) *MockOrganizationRepositoryPrimaryKeyConditionCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockOrganizationRepositoryPrimaryKeyConditionCall) DoAndReturn(f func(string, string) database.Condition) *MockOrganizationRepositoryPrimaryKeyConditionCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
 // SetName mocks base method.
 func (m *MockOrganizationRepository) SetName(name string) database.Change {
 	m.ctrl.T.Helper()

backend/v3/domain/mock/org_domain.mock.go

@@ -587,6 +587,82 @@ func (c *MockOrganizationDomainRepositoryOrgIDConditionCall) DoAndReturn(f func(
 	return c
 }
 
+// PrimaryKeyColumns mocks base method.
+func (m *MockOrganizationDomainRepository) PrimaryKeyColumns() []database.Column {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "PrimaryKeyColumns")
+	ret0, _ := ret[0].([]database.Column)
+	return ret0
+}
+
+// PrimaryKeyColumns indicates an expected call of PrimaryKeyColumns.
+func (mr *MockOrganizationDomainRepositoryMockRecorder) PrimaryKeyColumns() *MockOrganizationDomainRepositoryPrimaryKeyColumnsCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PrimaryKeyColumns", reflect.TypeOf((*MockOrganizationDomainRepository)(nil).PrimaryKeyColumns))
+	return &MockOrganizationDomainRepositoryPrimaryKeyColumnsCall{Call: call}
+}
+
+// MockOrganizationDomainRepositoryPrimaryKeyColumnsCall wrap *gomock.Call
+type MockOrganizationDomainRepositoryPrimaryKeyColumnsCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockOrganizationDomainRepositoryPrimaryKeyColumnsCall) Return(arg0 []database.Column) *MockOrganizationDomainRepositoryPrimaryKeyColumnsCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockOrganizationDomainRepositoryPrimaryKeyColumnsCall) Do(f func() []database.Column) *MockOrganizationDomainRepositoryPrimaryKeyColumnsCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockOrganizationDomainRepositoryPrimaryKeyColumnsCall) DoAndReturn(f func() []database.Column) *MockOrganizationDomainRepositoryPrimaryKeyColumnsCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
+// PrimaryKeyCondition mocks base method.
+func (m *MockOrganizationDomainRepository) PrimaryKeyCondition(instanceID, orgID, arg2 string) database.Condition {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "PrimaryKeyCondition", instanceID, orgID, arg2)
+	ret0, _ := ret[0].(database.Condition)
+	return ret0
+}
+
+// PrimaryKeyCondition indicates an expected call of PrimaryKeyCondition.
+func (mr *MockOrganizationDomainRepositoryMockRecorder) PrimaryKeyCondition(instanceID, orgID, arg2 any) *MockOrganizationDomainRepositoryPrimaryKeyConditionCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PrimaryKeyCondition", reflect.TypeOf((*MockOrganizationDomainRepository)(nil).PrimaryKeyCondition), instanceID, orgID, arg2)
+	return &MockOrganizationDomainRepositoryPrimaryKeyConditionCall{Call: call}
+}
+
+// MockOrganizationDomainRepositoryPrimaryKeyConditionCall wrap *gomock.Call
+type MockOrganizationDomainRepositoryPrimaryKeyConditionCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockOrganizationDomainRepositoryPrimaryKeyConditionCall) Return(arg0 database.Condition) *MockOrganizationDomainRepositoryPrimaryKeyConditionCall {
+	c.Call = c.Call.Return(arg0)
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockOrganizationDomainRepositoryPrimaryKeyConditionCall) Do(f func(string, string, string) database.Condition) *MockOrganizationDomainRepositoryPrimaryKeyConditionCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockOrganizationDomainRepositoryPrimaryKeyConditionCall) DoAndReturn(f func(string, string, string) database.Condition) *MockOrganizationDomainRepositoryPrimaryKeyConditionCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
 // Remove mocks base method.
 func (m *MockOrganizationDomainRepository) Remove(ctx context.Context, client database.QueryExecutor, condition database.Condition) (int64, error) {
 	m.ctrl.T.Helper()

backend/v3/domain/mock/transactional.mock.go

@@ -0,0 +1,52 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: github.com/zitadel/zitadel/backend/v3/domain (interfaces: Transactional)
+//
+// Generated by this command:
+//
+//	mockgen -destination=mock/transactional.mock.go -package=domainmock . Transactional
+//
+
+// Package domainmock is a generated GoMock package.
+package domainmock
+
+import (
+	reflect "reflect"
+
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockTransactional is a mock of Transactional interface.
+type MockTransactional struct {
+	ctrl     *gomock.Controller
+	recorder *MockTransactionalMockRecorder
+	isgomock struct{}
+}
+
+// MockTransactionalMockRecorder is the mock recorder for MockTransactional.
+type MockTransactionalMockRecorder struct {
+	mock *MockTransactional
+}
+
+// NewMockTransactional creates a new mock instance.
+func NewMockTransactional(ctrl *gomock.Controller) *MockTransactional {
+	mock := &MockTransactional{ctrl: ctrl}
+	mock.recorder = &MockTransactionalMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockTransactional) EXPECT() *MockTransactionalMockRecorder {
+	return m.recorder
+}
+
+// RequiresTransaction mocks base method.
+func (m *MockTransactional) RequiresTransaction() {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "RequiresTransaction")
+}
+
+// RequiresTransaction indicates an expected call of RequiresTransaction.
+func (mr *MockTransactionalMockRecorder) RequiresTransaction() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RequiresTransaction", reflect.TypeOf((*MockTransactional)(nil).RequiresTransaction))
+}

backend/v3/domain/org_activate.go

@@ -12,7 +12,10 @@ import (
 	"github.com/zitadel/zitadel/internal/zerrors"
 )
 
-var _ Commander = (*ActivateOrgCommand)(nil)
+var (
+	_ Commander     = (*ActivateOrgCommand)(nil)
+	_ Transactional = (*ActivateOrgCommand)(nil)
+)
 
 type ActivateOrgCommand struct {
 	ID string `json:"id"`
@@ -22,26 +25,20 @@ func NewActivateOrgCommand(organizationID string) *ActivateOrgCommand {
 	return &ActivateOrgCommand{ID: organizationID}
 }
 
+// RequiresTransaction implements [Transactional].
+func (cmd *ActivateOrgCommand) RequiresTransaction() {}
+
 // Events implements [Commander].
-func (d *ActivateOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]eventstore.Command, error) {
-	return []eventstore.Command{org.NewOrgReactivatedEvent(ctx, &org.NewAggregate(d.ID).Aggregate)}, nil
+func (cmd *ActivateOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]eventstore.Command, error) {
+	return []eventstore.Command{org.NewOrgReactivatedEvent(ctx, &org.NewAggregate(cmd.ID).Aggregate)}, nil
 }
 
 // Execute implements [Commander].
-func (d *ActivateOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err error) {
-	close, err := opts.EnsureTx(ctx)
-	if err != nil {
-		return err
-	}
-	defer func() { err = close(ctx, err) }()
-
+func (cmd *ActivateOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err error) {
 	organizationRepo := opts.organizationRepo
 
-	updateCount, err := organizationRepo.Update(ctx, pool,
-		database.And(
-			organizationRepo.IDCondition(d.ID),
-			organizationRepo.InstanceIDCondition(authz.GetInstance(ctx).InstanceID()),
-		),
+	updateCount, err := organizationRepo.Update(ctx, opts.DB(),
+		organizationRepo.PrimaryKeyCondition(authz.GetInstance(ctx).InstanceID(), cmd.ID),
 		database.NewChange(organizationRepo.StateColumn(), OrgStateActive),
 	)
 
@@ -62,24 +59,21 @@ func (d *ActivateOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err
 }
 
 // String implements [Commander].
-func (d *ActivateOrgCommand) String() string {
+func (cmd *ActivateOrgCommand) String() string {
 	return "ActivateOrgCommand"
 }
 
 // Validate implements [Commander].
-func (d *ActivateOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) (err error) {
-	if d.ID = strings.TrimSpace(d.ID); d.ID == "" {
+func (cmd *ActivateOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) (err error) {
+	if cmd.ID = strings.TrimSpace(cmd.ID); cmd.ID == "" {
 		return zerrors.ThrowInvalidArgument(nil, "DOM-hJuuAv", "invalid organization ID")
 	}
 
 	organizationRepo := opts.organizationRepo
 
 	// TODO: lock entry as soon as https://github.com/zitadel/zitadel/issues/10930 is done
-	org, err := organizationRepo.Get(ctx, pool, database.WithCondition(
-		database.And(
-			organizationRepo.IDCondition(d.ID),
-			organizationRepo.InstanceIDCondition(authz.GetInstance(ctx).InstanceID()),
-		),
+	org, err := organizationRepo.Get(ctx, opts.DB(), database.WithCondition(
+		organizationRepo.PrimaryKeyCondition(authz.GetInstance(ctx).InstanceID(), cmd.ID),
 	))
 	if err != nil {
 		var notFoundError *database.NoRowFoundError

backend/v3/domain/org_activate_test.go

@@ -63,10 +63,7 @@ func TestActivateOrgCommand_Validate(t *testing.T) {
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(
 						database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("instance-1"),
-							),
+							repo.PrimaryKeyCondition("instance-1", "org-1"),
 						))).
 					Times(1).
 					Return(nil, getErr)
@@ -82,10 +79,7 @@ func TestActivateOrgCommand_Validate(t *testing.T) {
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(
 						database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("instance-1"),
-							),
+							repo.PrimaryKeyCondition("instance-1", "org-1"),
 						))).
 					Times(1).
 					Return(nil, database.NewNoRowFoundError(getErr))
@@ -100,10 +94,7 @@ func TestActivateOrgCommand_Validate(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -121,10 +112,7 @@ func TestActivateOrgCommand_Validate(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -145,8 +133,9 @@ func TestActivateOrgCommand_Validate(t *testing.T) {
 			ctx := authz.NewMockContext("instance-1", "", "")
 			ctrl := gomock.NewController(t)
 			opts := &domain.InvokeOpts{
-				DB: new(noopdb.Pool),
+				Invoker: domain.NewTransactionInvoker(nil),
 			}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
 			if tc.orgRepo != nil {
 				domain.WithOrganizationRepo(tc.orgRepo(ctrl))(opts)
 			}
@@ -193,10 +182,7 @@ func TestActivateOrgCommand_Execute(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetState(domain.OrgStateActive),
 					).
 					Times(1).
@@ -213,10 +199,7 @@ func TestActivateOrgCommand_Execute(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetState(domain.OrgStateActive)).
 					Times(1).
 					Return(int64(0), nil)
@@ -230,10 +213,7 @@ func TestActivateOrgCommand_Execute(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetState(domain.OrgStateActive)).
 					Times(1).
 					Return(int64(2), nil)
@@ -248,10 +228,7 @@ func TestActivateOrgCommand_Execute(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetState(domain.OrgStateActive)).
 					Times(1).
 					Return(int64(1), nil)
@@ -272,17 +249,19 @@ func TestActivateOrgCommand_Execute(t *testing.T) {
 			}
 
 			opts := &domain.InvokeOpts{
-				DB: new(noopdb.Pool),
+				Invoker: domain.NewTransactionInvoker(nil),
 			}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
+
 			if tc.orgRepo != nil {
 				domain.WithOrganizationRepo(tc.orgRepo(ctrl))(opts)
 			}
 			if tc.queryExecutor != nil {
-				opts.DB = tc.queryExecutor(ctrl)
+				domain.WithQueryExecutor(tc.queryExecutor(ctrl))(opts)
 			}
 
 			// Test
-			err := cmd.Execute(ctx, opts)
+			err := opts.Invoke(ctx, cmd)
 
 			// Verify
 			assert.Equal(t, tc.expectedError, err)

backend/v3/domain/org_deactivate.go

@@ -12,7 +12,10 @@ import (
 	"github.com/zitadel/zitadel/internal/zerrors"
 )
 
-var _ Commander = (*DeactivateOrgCommand)(nil)
+var (
+	_ Commander     = (*DeactivateOrgCommand)(nil)
+	_ Transactional = (*DeactivateOrgCommand)(nil)
+)
 
 type DeactivateOrgCommand struct {
 	ID string `json:"id"`
@@ -23,62 +26,50 @@ func NewDeactivateOrgCommand(organizationID string) *DeactivateOrgCommand {
 }
 
 // Events implements [Commander].
-func (d *DeactivateOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]eventstore.Command, error) {
-	return []eventstore.Command{org.NewOrgDeactivatedEvent(ctx, &org.NewAggregate(d.ID).Aggregate)}, nil
+func (cmd *DeactivateOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]eventstore.Command, error) {
+	return []eventstore.Command{org.NewOrgDeactivatedEvent(ctx, &org.NewAggregate(cmd.ID).Aggregate)}, nil
 }
 
-// Execute implements [Commander].
-func (d *DeactivateOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err error) {
-	close, err := opts.EnsureTx(ctx)
-	if err != nil {
-		return err
-	}
-	defer func() { err = close(ctx, err) }()
+// RequiresTransaction implements [Transactional].
+func (cmd *DeactivateOrgCommand) RequiresTransaction() {}
 
+// Execute implements [Commander].
+func (cmd *DeactivateOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err error) {
 	organizationRepo := opts.organizationRepo
 
-	updateCount, err := organizationRepo.Update(ctx, pool,
-		database.And(
-			organizationRepo.IDCondition(d.ID),
-			organizationRepo.InstanceIDCondition(authz.GetInstance(ctx).InstanceID()),
-		),
+	updateCount, err := organizationRepo.Update(ctx, opts.DB(),
+		organizationRepo.PrimaryKeyCondition(authz.GetInstance(ctx).InstanceID(), cmd.ID),
 		database.NewChange(organizationRepo.StateColumn(), OrgStateInactive),
 	)
-
 	if err != nil {
 		return err
 	}
 
 	if updateCount == 0 {
-		err = zerrors.ThrowNotFound(nil, "DOM-vWPy7D", "Errors.Org.NotFound")
-		return err
+		return zerrors.ThrowNotFound(nil, "DOM-vWPy7D", "Errors.Org.NotFound")
 	}
 	if updateCount > 1 {
-		err = zerrors.ThrowInternal(NewMultipleObjectsUpdatedError(1, updateCount), "DOM-dXl1kJ", "unexpected number of rows updated")
-		return err
+		return zerrors.ThrowInternal(NewMultipleObjectsUpdatedError(1, updateCount), "DOM-dXl1kJ", "unexpected number of rows updated")
 	}
 
 	return nil
 }
 
 // String implements [Commander].
-func (d *DeactivateOrgCommand) String() string {
+func (DeactivateOrgCommand) String() string {
 	return "DeactivateOrgCommand"
 }
 
 // Validate implements [Commander].
-func (d *DeactivateOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) (err error) {
-	if d.ID = strings.TrimSpace(d.ID); d.ID == "" {
+func (cmd *DeactivateOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) (err error) {
+	if cmd.ID = strings.TrimSpace(cmd.ID); cmd.ID == "" {
 		return zerrors.ThrowInvalidArgument(nil, "DOM-Qc3T1r", "invalid organization ID")
 	}
 
 	organizationRepo := opts.organizationRepo
 
-	org, err := organizationRepo.Get(ctx, pool, database.WithCondition(
-		database.And(
-			organizationRepo.IDCondition(d.ID),
-			organizationRepo.InstanceIDCondition(authz.GetInstance(ctx).InstanceID()),
-		),
+	org, err := organizationRepo.Get(ctx, opts.DB(), database.WithCondition(
+		organizationRepo.PrimaryKeyCondition(authz.GetInstance(ctx).InstanceID(), cmd.ID),
 	))
 	if err != nil {
 		var notFoundError *database.NoRowFoundError

backend/v3/domain/org_deactivate_test.go

@@ -62,10 +62,7 @@ func TestDeactivateOrgCommand_Validate(t *testing.T) {
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(
 						database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("instance-1"),
-							),
+							repo.PrimaryKeyCondition("instance-1", "org-1"),
 						))).
 					Times(1).
 					Return(nil, getErr)
@@ -81,10 +78,7 @@ func TestDeactivateOrgCommand_Validate(t *testing.T) {
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(
 						database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("instance-1"),
-							),
+							repo.PrimaryKeyCondition("instance-1", "org-1"),
 						))).
 					Times(1).
 					Return(nil, database.NewNoRowFoundError(getErr))
@@ -99,10 +93,7 @@ func TestDeactivateOrgCommand_Validate(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -123,9 +114,8 @@ func TestDeactivateOrgCommand_Validate(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			cmd := domain.NewDeactivateOrgCommand(tc.inputOrgID)
 
-			opts := &domain.InvokeOpts{
-				DB: new(noopdb.Pool),
-			}
+			opts := &domain.InvokeOpts{}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
 			if tc.orgRepo != nil {
 				domain.WithOrganizationRepo(tc.orgRepo(ctrl))(opts)
 			}
@@ -141,7 +131,6 @@ func TestDeactivateOrgCommand_Validate(t *testing.T) {
 
 func TestDeactivateOrgCommand_Execute(t *testing.T) {
 	t.Parallel()
-	txInitErr := errors.New("tx init error")
 	updateErr := errors.New("update error")
 
 	tt := []struct {
@@ -154,28 +143,13 @@ func TestDeactivateOrgCommand_Execute(t *testing.T) {
 
 		expectedError error
 	}{
-		{
-			testName: "when EnsureTx fails should return error",
-			queryExecutor: func(ctrl *gomock.Controller) database.QueryExecutor {
-				mockDB := dbmock.NewMockPool(ctrl)
-				mockDB.EXPECT().
-					Begin(gomock.Any(), gomock.Any()).
-					Times(1).
-					Return(nil, txInitErr)
-				return mockDB
-			},
-			expectedError: txInitErr,
-		},
 		{
 			testName: "when org update fails should return error",
 			orgRepo: func(ctrl *gomock.Controller) domain.OrganizationRepository {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetState(domain.OrgStateInactive),
 					).
 					Times(1).
@@ -192,10 +166,7 @@ func TestDeactivateOrgCommand_Execute(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetState(domain.OrgStateInactive)).
 					Times(1).
 					Return(int64(0), nil)
@@ -209,10 +180,7 @@ func TestDeactivateOrgCommand_Execute(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetState(domain.OrgStateInactive)).
 					Times(1).
 					Return(int64(2), nil)
@@ -227,10 +195,7 @@ func TestDeactivateOrgCommand_Execute(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetState(domain.OrgStateInactive)).
 					Times(1).
 					Return(int64(1), nil)
@@ -250,18 +215,17 @@ func TestDeactivateOrgCommand_Execute(t *testing.T) {
 				ID: tc.inputID,
 			}
 
-			opts := &domain.InvokeOpts{
-				DB: new(noopdb.Pool),
-			}
+			opts := &domain.InvokeOpts{}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
 			if tc.orgRepo != nil {
 				domain.WithOrganizationRepo(tc.orgRepo(ctrl))(opts)
 			}
 			if tc.queryExecutor != nil {
-				opts.DB = tc.queryExecutor(ctrl)
+				domain.WithQueryExecutor(tc.queryExecutor(ctrl))(opts)
 			}
 
 			// Test
-			err := cmd.Execute(ctx, opts)
+			err := opts.Invoke(ctx, cmd)
 
 			// Verify
 			assert.Equal(t, tc.expectedError, err)

backend/v3/domain/org_delete.go

@@ -23,20 +23,17 @@ func NewDeleteOrgCommand(organizationID string) *DeleteOrgCommand {
 	return &DeleteOrgCommand{ID: organizationID}
 }
 
+// RequiresTransaction implements [Transactional].
+func (cmd *DeleteOrgCommand) RequiresTransaction() {}
+
 // Events implements [Commander].
 //
 // TODO(IAM-Marco): Finish implementation when policies, org settings, idp links and entities repositories
 // are implemented
-func (d *DeleteOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]eventstore.Command, error) {
-	closeFunc, err := opts.EnsureTx(ctx)
-	if err != nil {
-		return nil, err
-	}
-	defer func() { err = closeFunc(ctx, err) }()
-
+func (cmd *DeleteOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]eventstore.Command, error) {
 	usernames := []string{}
-	// userRepo := opts.usersRepo(pool)
-	// users, err := userRepo.List(ctx, database.WithCondition(opts.organizationRepo(pool).IDCondition(d.ID)))
+	// userRepo := opts.usersRepo(opts.DB())
+	// users, err := userRepo.List(ctx, database.WithCondition(opts.organizationRepo(opts.DB()).IDCondition(d.ID)))
 	// if err != nil {
 	// 	return nil, err
 	// }
@@ -44,13 +41,13 @@ func (d *DeleteOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]even
 	// 	usernames = append(usernames, u.UserID)
 	// }
 
-	// domainPolicyRepo := opts.domainPolicyRepo(pool)
+	// domainPolicyRepo := opts.domainPolicyRepo(opts.DB())
 	// policy, err := domainPolicyRepo.Get(ctx, instanceID, d.ID)
 	// if err != nil {
 	// 	return nil, err
 	// }
 
-	// orgSettingsRepo := opts.organizationSettingsRepo(pool)
+	// orgSettingsRepo := opts.organizationSettingsRepo(opts.DB())
 	// orgSettings, err := orgSettingsRepo.Get(ctx, d.ID)
 	// if err != nil {
 	// 	return nil, err
@@ -58,13 +55,13 @@ func (d *DeleteOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]even
 	// areUsernamesOrganizationScoped := policy.UserLoginMustBeDomain || orgSettings.UsernamesUnique
 	areUsernamesOrganizationScoped := false
 
-	domainNames := make([]string, len(d.Domains))
-	for i, domain := range d.Domains {
+	domainNames := make([]string, len(cmd.Domains))
+	for i, domain := range cmd.Domains {
 		domainNames[i] = domain.Domain
 	}
 
 	externalIDPLinks := []*internal_domain.UserIDPLink{}
-	// idpLinksRepo := opts.idpLinksRepo(pool)
+	// idpLinksRepo := opts.idpLinksRepo(opts.DB())
 	// idpLinks, err := idpLinksRepo.List(ctx, d.ID)
 	// if err != nil {
 	// 	return nil, err
@@ -75,7 +72,7 @@ func (d *DeleteOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]even
 	// }
 
 	samlEntityIDs := []string{}
-	// entityIDsRepo := opts.entityIDsRepo(pool)
+	// entityIDsRepo := opts.entityIDsRepo(opts.DB())
 	// entityIDs, err := entityIDsRepo.List(ctx, d.ID)
 	// if err != nil {
 	// 	return nil, err
@@ -87,8 +84,8 @@ func (d *DeleteOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]even
 	return []eventstore.Command{
 		org.NewOrgRemovedEvent(
 			ctx,
-			&org.NewAggregate(d.ID).Aggregate,
-			d.OrganizationName,
+			&org.NewAggregate(cmd.ID).Aggregate,
+			cmd.OrganizationName,
 			usernames,
 			areUsernamesOrganizationScoped,
 			domainNames,
@@ -99,32 +96,22 @@ func (d *DeleteOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]even
 }
 
 // Execute implements [Commander].
-func (d *DeleteOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err error) {
-	closeFunc, err := opts.EnsureTx(ctx)
-	if err != nil {
-		return err
-	}
-
-	defer func() { err = closeFunc(ctx, err) }()
+func (cmd *DeleteOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err error) {
 	instance := authz.GetInstance(ctx)
 
 	orgRepo := opts.organizationRepo.LoadDomains()
 
-	orgToDelete, err := orgRepo.Get(ctx, pool, database.WithCondition(database.And(
-		orgRepo.IDCondition(d.ID),
-		orgRepo.InstanceIDCondition(instance.InstanceID()),
-	)))
+	orgToDelete, err := orgRepo.Get(ctx, opts.DB(), database.WithCondition(
+		orgRepo.PrimaryKeyCondition(instance.InstanceID(), cmd.ID),
+	))
 	if err != nil {
 		return err
 	}
-	d.OrganizationName = orgToDelete.Name
-	d.Domains = orgToDelete.Domains
+	cmd.OrganizationName = orgToDelete.Name
+	cmd.Domains = orgToDelete.Domains
 
-	deletedRows, err := orgRepo.Delete(ctx, pool,
-		database.And(
-			orgRepo.IDCondition(d.ID),
-			orgRepo.InstanceIDCondition(authz.GetInstance(ctx).InstanceID()),
-		),
+	deletedRows, err := orgRepo.Delete(ctx, opts.DB(),
+		orgRepo.PrimaryKeyCondition(authz.GetInstance(ctx).InstanceID(), cmd.ID),
 	)
 	if err != nil {
 		return err
@@ -142,25 +129,25 @@ func (d *DeleteOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err e
 }
 
 // String implements [Commander].
-func (d *DeleteOrgCommand) String() string {
+func (DeleteOrgCommand) String() string {
 	return "DeleteOrgCommand"
 }
 
 // Validate implements [Commander].
-func (d *DeleteOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) (err error) {
+func (cmd *DeleteOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) (err error) {
 	instance := authz.GetInstance(ctx)
 
-	d.ID = strings.TrimSpace(d.ID)
-	if d.ID == instance.DefaultOrganisationID() {
+	cmd.ID = strings.TrimSpace(cmd.ID)
+	if cmd.ID == instance.DefaultOrganisationID() {
 		return zerrors.ThrowPreconditionFailed(nil, "DOM-LCkE69", "Errors.Org.DefaultOrgNotDeletable")
 	}
 
 	// Check if the ZITADEL project exists on the input organization
 	projectRepo := opts.projectRepo
-	_, getErr := projectRepo.Get(ctx, pool,
+	_, getErr := projectRepo.Get(ctx, opts.DB(),
 		database.WithCondition(database.And(
 			projectRepo.IDCondition(instance.ProjectID()),
-			projectRepo.OrganizationIDCondition(d.ID),
+			projectRepo.OrganizationIDCondition(cmd.ID),
 			projectRepo.InstanceIDCondition(instance.InstanceID()),
 		)),
 	)
@@ -174,11 +161,10 @@ func (d *DeleteOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) (err
 	}
 
 	orgRepo := opts.organizationRepo
-	_, errGetOrg := orgRepo.Get(ctx, pool,
-		database.WithCondition(database.And(
-			orgRepo.IDCondition(d.ID),
-			orgRepo.InstanceIDCondition(instance.InstanceID()),
-		)))
+	_, errGetOrg := orgRepo.Get(ctx, opts.DB(),
+		database.WithCondition(
+			orgRepo.PrimaryKeyCondition(instance.InstanceID(), cmd.ID),
+		))
 	if errGetOrg != nil {
 		if errors.Is(errGetOrg, &database.NoRowFoundError{}) {
 			err = zerrors.ThrowNotFound(errGetOrg, "DOM-8KYOH3", "Errors.Org.NotFound")
@@ -189,4 +175,7 @@ func (d *DeleteOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) (err
 	return err
 }
 
-var _ Commander = (*DeleteOrgCommand)(nil)
+var (
+	_ Commander     = (*DeleteOrgCommand)(nil)
+	_ Transactional = (*DeleteOrgCommand)(nil)
+)

backend/v3/domain/org_delete_test.go

@@ -100,7 +100,7 @@ func TestDeleteOrgCommand_Validate(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(repo.IDCondition("org-1"), repo.InstanceIDCondition("inst-1")),
+						repo.PrimaryKeyCondition("inst-1", "org-1"),
 					))).
 					Times(1).
 					Return(nil, database.NewNoRowFoundError(getErr))
@@ -131,7 +131,7 @@ func TestDeleteOrgCommand_Validate(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(repo.IDCondition("org-1"), repo.InstanceIDCondition("inst-1")),
+						repo.PrimaryKeyCondition("inst-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -164,7 +164,7 @@ func TestDeleteOrgCommand_Validate(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(repo.IDCondition("org-1"), repo.InstanceIDCondition("inst-1")),
+						repo.PrimaryKeyCondition("inst-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -182,7 +182,8 @@ func TestDeleteOrgCommand_Validate(t *testing.T) {
 			// Given
 			d := domain.NewDeleteOrgCommand(tc.inputOrganizationID)
 			ctrl := gomock.NewController(t)
-			opts := &domain.InvokeOpts{DB: new(noopdb.Pool)}
+			opts := &domain.InvokeOpts{}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
 
 			if tc.orgRepo != nil {
 				domain.WithOrganizationRepo(tc.orgRepo(ctrl))(opts)
@@ -204,7 +205,6 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 	t.Parallel()
 
 	ctx := authz.NewMockContext("inst-1", "org-1", gofakeit.UUID())
-	txInitErr := errors.New("tx init error")
 	deleteErr := errors.New("delete error")
 	getErr := errors.New("get error")
 
@@ -218,18 +218,6 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 		expectedError   error
 		expectedOrgName string
 	}{
-		{
-			testName: "when EnsureTx fails should return error",
-			mockTx: func(ctrl *gomock.Controller) database.QueryExecutor {
-				mockDB := dbmock.NewMockPool(ctrl)
-				mockDB.EXPECT().
-					Begin(gomock.Any(), gomock.Any()).
-					Times(1).
-					Return(nil, txInitErr)
-				return mockDB
-			},
-			expectedError: txInitErr,
-		},
 		{
 			testName: "when retrieving organization fails should return error",
 			orgRepo: func(ctrl *gomock.Controller) domain.OrganizationRepository {
@@ -245,10 +233,7 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 						gomock.Any(),
 						gomock.Any(),
 						dbmock.QueryOptions(database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("inst-1"),
-							),
+							repo.PrimaryKeyCondition("inst-1", "org-1"),
 						)),
 					).
 					Times(1).
@@ -273,10 +258,7 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 						gomock.Any(),
 						gomock.Any(),
 						dbmock.QueryOptions(database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("inst-1"),
-							),
+							repo.PrimaryKeyCondition("inst-1", "org-1"),
 						)),
 					).
 					Times(1).
@@ -286,9 +268,7 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 					}, nil)
 				repo.EXPECT().
 					Delete(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("inst-1")),
+						repo.PrimaryKeyCondition("inst-1", "org-1"),
 					).
 					Times(1).
 					Return(int64(0), deleteErr)
@@ -313,10 +293,7 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 						gomock.Any(),
 						gomock.Any(),
 						dbmock.QueryOptions(database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("inst-1"),
-							),
+							repo.PrimaryKeyCondition("inst-1", "org-1"),
 						)),
 					).
 					Times(1).
@@ -325,10 +302,7 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 						Name: "organization 1",
 					}, nil)
 				repo.EXPECT().
-					Delete(gomock.Any(), gomock.Any(), database.And(
-						repo.IDCondition("org-1"),
-						repo.InstanceIDCondition("inst-1")),
-					).
+					Delete(gomock.Any(), gomock.Any(), repo.PrimaryKeyCondition("inst-1", "org-1")).
 					Times(1).
 					Return(int64(2), nil)
 				return repo
@@ -352,10 +326,7 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 						gomock.Any(),
 						gomock.Any(),
 						dbmock.QueryOptions(database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("inst-1"),
-							),
+							repo.PrimaryKeyCondition("inst-1", "org-1"),
 						)),
 					).
 					Times(1).
@@ -364,10 +335,7 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 						Name: "organization 1",
 					}, nil)
 				repo.EXPECT().
-					Delete(gomock.Any(), gomock.Any(), database.And(
-						repo.IDCondition("org-1"),
-						repo.InstanceIDCondition("inst-1")),
-					).
+					Delete(gomock.Any(), gomock.Any(), repo.PrimaryKeyCondition("inst-1", "org-1")).
 					Times(1).
 					Return(int64(0), nil)
 				return repo
@@ -391,10 +359,7 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 						gomock.Any(),
 						gomock.Any(),
 						dbmock.QueryOptions(database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("inst-1"),
-							),
+							repo.PrimaryKeyCondition("inst-1", "org-1"),
 						)),
 					).
 					Times(1).
@@ -403,10 +368,7 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 						Name: "organization 1",
 					}, nil)
 				repo.EXPECT().
-					Delete(gomock.Any(), gomock.Any(), database.And(
-						repo.IDCondition("org-1"),
-						repo.InstanceIDCondition("inst-1")),
-					).
+					Delete(gomock.Any(), gomock.Any(), repo.PrimaryKeyCondition("inst-1", "org-1")).
 					Times(1).
 					Return(int64(1), nil)
 				return repo
@@ -419,23 +381,23 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 		t.Run(tc.testName, func(t *testing.T) {
 			t.Parallel()
 			// Given
-			d := domain.NewDeleteOrgCommand(tc.inputOrganizationID)
+			cmd := domain.NewDeleteOrgCommand(tc.inputOrganizationID)
 			ctrl := gomock.NewController(t)
-			opts := &domain.InvokeOpts{DB: new(noopdb.Pool)}
-
+			opts := &domain.InvokeOpts{}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
 			if tc.mockTx != nil {
-				opts.DB = tc.mockTx(ctrl)
+				domain.WithQueryExecutor(tc.mockTx(ctrl))(opts)
 			}
 			if tc.orgRepo != nil {
 				domain.WithOrganizationRepo(tc.orgRepo(ctrl))(opts)
 			}
 
 			// Test
-			err := d.Execute(ctx, opts)
+			err := opts.Invoke(ctx, cmd)
 
 			// Verify
 			assert.Equal(t, tc.expectedError, err)
-			assert.Equal(t, tc.expectedOrgName, d.OrganizationName)
+			assert.Equal(t, tc.expectedOrgName, cmd.OrganizationName)
 		})
 	}
 }
@@ -444,7 +406,6 @@ func TestDeleteOrgCommand_Execute(t *testing.T) {
 func TestDeleteOrgCommand_Events(t *testing.T) {
 	t.Parallel()
 	ctx := authz.NewMockContext("inst-1", "org-1", gofakeit.UUID())
-	txInitErr := errors.New("tx init error")
 
 	tt := []struct {
 		testName      string
@@ -453,19 +414,6 @@ func TestDeleteOrgCommand_Events(t *testing.T) {
 		expectedError error
 		expectedCount int
 	}{
-		{
-			testName: "when EnsureTx fails should return error",
-			mockTx: func(ctrl *gomock.Controller) database.QueryExecutor {
-				mockDB := dbmock.NewMockPool(ctrl)
-				mockDB.EXPECT().
-					Begin(gomock.Any(), gomock.Any()).
-					Times(1).
-					Return(nil, txInitErr)
-				return mockDB
-			},
-			command:       domain.NewDeleteOrgCommand("org-1"),
-			expectedError: txInitErr,
-		},
 		{
 			testName: "should create org removed event",
 			command: &domain.DeleteOrgCommand{
@@ -486,10 +434,11 @@ func TestDeleteOrgCommand_Events(t *testing.T) {
 
 			// Given
 			ctrl := gomock.NewController(t)
-			opts := &domain.InvokeOpts{DB: new(noopdb.Pool)}
+			opts := &domain.InvokeOpts{}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
 
 			if tc.mockTx != nil {
-				opts.DB = tc.mockTx(ctrl)
+				domain.WithQueryExecutor(tc.mockTx(ctrl))(opts)
 			}
 
 			// Test

backend/v3/domain/org_list.go

@@ -40,7 +40,7 @@ func (l *ListOrgsQuery) Execute(ctx context.Context, opts *InvokeOpts) (err erro
 		return err
 	}
 
-	l.result, err = organizationRepo.List(ctx, pool, conditions, sorting, limit, pagination)
+	l.result, err = organizationRepo.List(ctx, opts.DB(), conditions, sorting, limit, pagination)
 	return err
 }
 

backend/v3/domain/org_list_test.go

@@ -317,9 +317,8 @@ func TestListOrgsCommand_Execute(t *testing.T) {
 			cmd := &domain.ListOrgsQuery{
 				Request: tc.inputRequest,
 			}
-			opts := &domain.InvokeOpts{
-				DB: new(noopdb.Pool),
-			}
+			opts := &domain.InvokeOpts{}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
 			if tc.repos != nil {
 				orgRepo, domainRepo := tc.repos(ctrl, tc.queryParams...)
 				domain.WithOrganizationRepo(orgRepo)(opts)
@@ -327,7 +326,7 @@ func TestListOrgsCommand_Execute(t *testing.T) {
 			}
 
 			// Test
-			err := cmd.Execute(ctx, opts)
+			err := opts.Invoke(ctx, cmd)
 
 			// Verify
 			assert.Equal(t, tc.expectedError, err)

backend/v3/domain/org_update.go

@@ -21,18 +21,24 @@ type UpdateOrgCommand struct {
 	IsOldDomainVerified *bool   `json:"is_old_domain_primary"`
 }
 
+// RequiresTransaction implements [Transactional].
+func (cmd *UpdateOrgCommand) RequiresTransaction() {}
+
 // Events implements Commander.
-func (u *UpdateOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]eventstore.Command, error) {
+func (cmd *UpdateOrgCommand) Events(ctx context.Context, opts *InvokeOpts) ([]eventstore.Command, error) {
 	toReturn := []eventstore.Command{}
 
-	if u.OldDomainName != nil && *u.OldDomainName != u.Name {
-		toReturn = append(toReturn, org.NewOrgChangedEvent(ctx, &org.NewAggregate(u.ID).Aggregate, *u.OldDomainName, u.Name))
+	if cmd.OldDomainName != nil && *cmd.OldDomainName != cmd.Name {
+		toReturn = append(toReturn, org.NewOrgChangedEvent(ctx, &org.NewAggregate(cmd.ID).Aggregate, *cmd.OldDomainName, cmd.Name))
 	}
 
 	return toReturn, nil
 }
 
-var _ Commander = (*UpdateOrgCommand)(nil)
+var (
+	_ Commander     = (*UpdateOrgCommand)(nil)
+	_ Transactional = (*UpdateOrgCommand)(nil)
+)
 
 func NewUpdateOrgCommand(id, name string) *UpdateOrgCommand {
 	return &UpdateOrgCommand{
@@ -41,38 +47,26 @@ func NewUpdateOrgCommand(id, name string) *UpdateOrgCommand {
 	}
 }
 
-func (u *UpdateOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err error) {
-	close, err := opts.EnsureTx(ctx)
-	if err != nil {
-		return err
-	}
-	defer func() { err = close(ctx, err) }()
-
+func (cmd *UpdateOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err error) {
 	organizationRepo := opts.organizationRepo.LoadDomains()
 
-	org, err := organizationRepo.Get(ctx, pool, database.WithCondition(
-		database.And(
-			organizationRepo.IDCondition(u.ID),
-			organizationRepo.InstanceIDCondition(authz.GetInstance(ctx).InstanceID()),
-		),
+	org, err := organizationRepo.Get(ctx, opts.DB(), database.WithCondition(
+		organizationRepo.PrimaryKeyCondition(authz.GetInstance(ctx).InstanceID(), cmd.ID),
 	))
 	if err != nil {
 		return err
 	}
 
-	err = u.setDomainInfos(ctx, org)
+	err = cmd.setDomainInfos(ctx, org)
 	if err != nil {
 		return err
 	}
 
 	updateCount, err := organizationRepo.Update(
 		ctx,
-		pool,
-		database.And(
-			organizationRepo.IDCondition(org.ID),
-			organizationRepo.InstanceIDCondition(org.InstanceID),
-		),
-		database.NewChange(organizationRepo.NameColumn(), u.Name),
+		opts.DB(),
+		organizationRepo.PrimaryKeyCondition(org.InstanceID, org.ID),
+		database.NewChange(organizationRepo.NameColumn(), cmd.Name),
 	)
 	if err != nil {
 		return err
@@ -90,45 +84,36 @@ func (u *UpdateOrgCommand) Execute(ctx context.Context, opts *InvokeOpts) (err e
 	return err
 }
 
-func (u *UpdateOrgCommand) String() string {
+func (cmd *UpdateOrgCommand) String() string {
 	return "UpdateOrgCommand"
 }
 
-func (u *UpdateOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) error {
-	if u.ID = strings.TrimSpace(u.ID); u.ID == "" {
+func (cmd *UpdateOrgCommand) Validate(ctx context.Context, opts *InvokeOpts) error {
+	if cmd.ID = strings.TrimSpace(cmd.ID); cmd.ID == "" {
 		return zerrors.ThrowInvalidArgument(nil, "DOM-lEMhVC", "invalid organization ID")
 	}
-	if u.Name = strings.TrimSpace(u.Name); u.Name == "" {
+	if cmd.Name = strings.TrimSpace(cmd.Name); cmd.Name == "" {
 		return zerrors.ThrowInvalidArgument(nil, "DOM-wfUntW", "invalid organization name")
 	}
 
-	close, err := opts.EnsureTx(ctx)
-	if err != nil {
-		return err
-	}
-	defer func() { err = close(ctx, err) }()
-
 	organizationRepo := opts.organizationRepo
 
-	org, err := organizationRepo.Get(ctx, pool, database.WithCondition(
-		database.And(
-			organizationRepo.IDCondition(u.ID),
-			organizationRepo.InstanceIDCondition(authz.GetInstance(ctx).InstanceID()),
-		),
-	))
+	org, err := organizationRepo.Get(ctx, opts.DB(),
+		database.WithCondition(
+			opts.organizationRepo.PrimaryKeyCondition(authz.GetInstance(ctx).InstanceID(), cmd.ID),
+		))
 	if err != nil {
 		return err
 	}
 
-	if org.Name == u.Name {
-		err = zerrors.ThrowPreconditionFailed(nil, "DOM-nDzwIu", "Errors.Org.NotChanged")
-		return err
+	if org.Name == cmd.Name {
+		return zerrors.ThrowPreconditionFailed(nil, "DOM-nDzwIu", "Errors.Org.NotChanged")
 	}
 
 	return nil
 }
 
-func (u *UpdateOrgCommand) setDomainInfos(ctx context.Context, org *Organization) error {
+func (cmd *UpdateOrgCommand) setDomainInfos(ctx context.Context, org *Organization) error {
 	iamDomain := http_utils.DomainContext(ctx).RequestedDomain()
 	oldDomainName, err := domain.NewIAMDomainName(org.Name, iamDomain)
 	if err != nil {
@@ -137,8 +122,8 @@ func (u *UpdateOrgCommand) setDomainInfos(ctx context.Context, org *Organization
 
 	for _, d := range org.Domains {
 		if d.Domain == oldDomainName && !d.IsPrimary {
-			u.IsOldDomainVerified = &d.IsVerified
-			u.OldDomainName = &d.Domain
+			cmd.IsOldDomainVerified = &d.IsVerified
+			cmd.OldDomainName = &d.Domain
 			return nil
 		}
 	}

backend/v3/domain/org_update_test.go

@@ -62,10 +62,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(
 						database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("instance-1"),
-							),
+							repo.PrimaryKeyCondition("instance-1", "org-1"),
 						))).
 					Times(1).
 					Return(nil, getErr)
@@ -85,10 +82,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 					Return(repo)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -117,10 +111,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 					Return(repo)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -136,10 +127,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetName("test org update"),
 					).
 					Times(1).
@@ -162,10 +150,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 					Return(repo)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -182,10 +167,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetName("test org update")).
 					Times(1).
 					Return(int64(0), nil)
@@ -205,10 +187,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 					Return(repo)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -225,10 +204,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetName("test org update")).
 					Times(1).
 					Return(int64(2), nil)
@@ -250,10 +226,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 					Return(repo)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -270,10 +243,7 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 
 				repo.EXPECT().
 					Update(gomock.Any(), gomock.Any(),
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 						repo.SetName("test org update")).
 					Times(1).
 					Return(int64(1), nil)
@@ -296,17 +266,19 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 			cmd := domain.NewUpdateOrgCommand(tc.inputID, tc.inputName)
 
 			opts := &domain.InvokeOpts{
-				DB: new(noopdb.Pool),
+				Invoker: domain.NewTransactionInvoker(nil),
 			}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
+
 			if tc.orgRepo != nil {
 				domain.WithOrganizationRepo(tc.orgRepo(ctrl))(opts)
 			}
 			if tc.queryExecutor != nil {
-				opts.DB = tc.queryExecutor(ctrl)
+				domain.WithQueryExecutor(tc.queryExecutor(ctrl))(opts)
 			}
 
 			// Test
-			err := cmd.Execute(ctx, opts)
+			err := opts.Invoke(ctx, cmd)
 
 			// Verify
 			assert.Equal(t, tc.expectedError, err)
@@ -318,7 +290,6 @@ func TestUpdateOrgCommand_Execute(t *testing.T) {
 
 func TestUpdateOrgCommand_Validate(t *testing.T) {
 	t.Parallel()
-	txInitErr := errors.New("tx init error")
 	getErr := errors.New("get error")
 
 	tt := []struct {
@@ -336,25 +307,11 @@ func TestUpdateOrgCommand_Validate(t *testing.T) {
 			expectedError: zerrors.ThrowInvalidArgument(nil, "DOM-lEMhVC", "invalid organization ID"),
 		},
 		{
-			testName:      "when no name shuld return invalid argument error",
+			testName:      "when no name should return invalid argument error",
 			inputOrgID:    "test-id",
 			inputOrgName:  "",
 			expectedError: zerrors.ThrowInvalidArgument(nil, "DOM-wfUntW", "invalid organization name"),
 		},
-		{
-			testName: "when EnsureTx fails should return error",
-			queryExecutor: func(ctrl *gomock.Controller) database.QueryExecutor {
-				mockDB := dbmock.NewMockPool(ctrl)
-				mockDB.EXPECT().
-					Begin(gomock.Any(), gomock.Any()).
-					Times(1).
-					Return(nil, txInitErr)
-				return mockDB
-			},
-			inputOrgID:    "test-id",
-			inputOrgName:  "test-name",
-			expectedError: txInitErr,
-		},
 		{
 			testName: "when retrieving org fails should return error",
 			orgRepo: func(ctrl *gomock.Controller) domain.OrganizationRepository {
@@ -362,10 +319,7 @@ func TestUpdateOrgCommand_Validate(t *testing.T) {
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(
 						database.WithCondition(
-							database.And(
-								repo.IDCondition("org-1"),
-								repo.InstanceIDCondition("instance-1"),
-							),
+							repo.PrimaryKeyCondition("instance-1", "org-1"),
 						))).
 					Times(1).
 					Return(nil, getErr)
@@ -381,10 +335,7 @@ func TestUpdateOrgCommand_Validate(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -403,10 +354,7 @@ func TestUpdateOrgCommand_Validate(t *testing.T) {
 				repo := domainmock.NewOrgRepo(ctrl)
 				repo.EXPECT().
 					Get(gomock.Any(), gomock.Any(), dbmock.QueryOptions(database.WithCondition(
-						database.And(
-							repo.IDCondition("org-1"),
-							repo.InstanceIDCondition("instance-1"),
-						),
+						repo.PrimaryKeyCondition("instance-1", "org-1"),
 					))).
 					Times(1).
 					Return(&domain.Organization{
@@ -426,19 +374,21 @@ func TestUpdateOrgCommand_Validate(t *testing.T) {
 			// Given
 			ctx := authz.NewMockContext("instance-1", "", "")
 			ctrl := gomock.NewController(t)
+
 			cmd := domain.NewUpdateOrgCommand(tc.inputOrgID, tc.inputOrgName)
 
 			opts := &domain.InvokeOpts{
-				DB: new(noopdb.Pool),
+				Invoker: domain.NewTransactionInvoker(domain.NewValidatorInvoker(nil)),
 			}
+			domain.WithQueryExecutor(new(noopdb.Pool))(opts)
 			if tc.orgRepo != nil {
 				domain.WithOrganizationRepo(tc.orgRepo(ctrl))(opts)
 			}
 			if tc.queryExecutor != nil {
-				opts.DB = tc.queryExecutor(ctrl)
+				domain.WithQueryExecutor(tc.queryExecutor(ctrl))(opts)
 			}
-
 			err := cmd.Validate(ctx, opts)
+
 			assert.Equal(t, tc.expectedError, err)
 		})
 	}

backend/v3/storage/database/condition.go

@@ -250,7 +250,7 @@ func NewTextIgnoreCaseCondition[T Text](col Column, op TextOperation, value T) C
 	}
 }
 
-// NewDateCondition creates a condition that compares a numeric column with a value.
+// NewNumberCondition creates a condition that compares a numeric column with a value.
 func NewNumberCondition[V Number](col Column, op NumberOperation, value V) Condition {
 	return valueCondition{
 		col: col,
@@ -260,7 +260,7 @@ func NewNumberCondition[V Number](col Column, op NumberOperation, value V) Condi
 	}
 }
 
-// NewDateCondition creates a condition that compares a boolean column with a value.
+// NewBooleanCondition creates a condition that compares a boolean column with a value.
 func NewBooleanCondition[V Boolean](col Column, value V) Condition {
 	return valueCondition{
 		col: col,
@@ -319,7 +319,7 @@ func (e *existsCondition) Matches(x any) bool {
 
 // String implements [Condition].
 func (e *existsCondition) String() string {
-	return "unimplemented"
+	return "existsCondition"
 }
 
 // Exists creates a condition that checks for the existence of rows in a subquery.

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

@@ -39,10 +39,7 @@ func TestServer_TestOrgDomainReduces(t *testing.T) {
 	assert.EventuallyWithT(t, func(t *assert.CollectT) {
 		_, err := orgRepo.Get(CTX, pool,
 			database.WithCondition(
-				database.And(
-					orgRepo.InstanceIDCondition(Instance.Instance.Id),
-					orgRepo.IDCondition(org.GetId()),
-				),
+				orgRepo.PrimaryKeyCondition(Instance.Instance.Id, org.GetId()),
 			),
 		)
 		assert.NoError(t, err)

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

@@ -1,55 +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/storage/database"
-	"github.com/zitadel/zitadel/backend/v3/storage/database/repository"
-	"github.com/zitadel/zitadel/internal/integration"
-	org "github.com/zitadel/zitadel/pkg/grpc/org/v2beta"
-	v2beta "github.com/zitadel/zitadel/pkg/grpc/org/v2beta"
-)
-
-func createOrg(t *testing.T) *org.CreateOrganizationResponse {
-	t.Helper()
-	org, err := OrgClient.CreateOrganization(CTX, &org.CreateOrganizationRequest{
-		Name: gofakeit.Name(),
-	})
-	require.NoError(t, err)
-
-	orgRepo := repository.OrganizationRepository()
-	retryDuration, tick := integration.WaitForAndTickWithMaxDuration(t.Context(), time.Minute)
-	assert.EventuallyWithT(t, func(tc *assert.CollectT) {
-		_, err := orgRepo.Get(t.Context(), pool,
-			database.WithCondition(database.And(
-				orgRepo.InstanceIDCondition(Instance.Instance.Id),
-				orgRepo.IDCondition(org.GetId())),
-			),
-		)
-		assert.NoError(tc, err)
-	}, retryDuration, tick)
-
-	return org
-}
-
-func createTestScopedOrg(t *testing.T) *org.CreateOrganizationResponse {
-	org := createOrg(t)
-
-	t.Cleanup(func() {
-		_, err := OrgClient.DeleteOrganization(CTX, &v2beta.DeleteOrganizationRequest{
-			Id: org.GetId(),
-		})
-		if err != nil {
-			t.Logf("Failed to delete organization on cleanup: %v", err)
-		}
-	})
-
-	return org
-}

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

@@ -14,6 +14,8 @@ import (
 	"github.com/zitadel/zitadel/backend/v3/storage/database"
 	"github.com/zitadel/zitadel/backend/v3/storage/database/repository"
 	"github.com/zitadel/zitadel/internal/integration"
+	org "github.com/zitadel/zitadel/pkg/grpc/org/v2beta"
+	v2beta "github.com/zitadel/zitadel/pkg/grpc/org/v2beta"
 	v2beta_org "github.com/zitadel/zitadel/pkg/grpc/org/v2beta"
 )
 
@@ -43,12 +45,7 @@ func TestServer_TestOrganizationReduces(t *testing.T) {
 		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
 		assert.EventuallyWithT(t, func(tt *assert.CollectT) {
 			organization, err := orgRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						orgRepo.IDCondition(org.GetId()),
-						orgRepo.InstanceIDCondition(instanceID),
-					),
-				),
+				database.WithCondition(orgRepo.PrimaryKeyCondition(instanceID, org.GetId())),
 			)
 			require.NoError(tt, err)
 
@@ -93,12 +90,7 @@ func TestServer_TestOrganizationReduces(t *testing.T) {
 		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
 		assert.EventuallyWithT(t, func(t *assert.CollectT) {
 			organization, err := orgRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						orgRepo.IDCondition(organization.Id),
-						orgRepo.InstanceIDCondition(instanceID),
-					),
-				),
+				database.WithCondition(orgRepo.PrimaryKeyCondition(instanceID, organization.GetId())),
 			)
 			require.NoError(t, err)
 
@@ -138,12 +130,7 @@ func TestServer_TestOrganizationReduces(t *testing.T) {
 		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
 		assert.EventuallyWithT(t, func(t *assert.CollectT) {
 			organization, err := orgRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						orgRepo.IDCondition(organization.Id),
-						orgRepo.InstanceIDCondition(instanceID),
-					),
-				),
+				database.WithCondition(orgRepo.PrimaryKeyCondition(instanceID, organization.GetId())),
 			)
 			require.NoError(t, err)
 
@@ -182,12 +169,7 @@ func TestServer_TestOrganizationReduces(t *testing.T) {
 		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
 		assert.EventuallyWithT(t, func(t *assert.CollectT) {
 			organization, err := orgRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						orgRepo.IDCondition(organization.Id),
-						orgRepo.InstanceIDCondition(instanceID),
-					),
-				),
+				database.WithCondition(orgRepo.PrimaryKeyCondition(instanceID, organization.GetId())),
 			)
 			require.NoError(t, err)
 			assert.Equal(t, domain.OrgStateInactive, organization.State)
@@ -204,12 +186,7 @@ func TestServer_TestOrganizationReduces(t *testing.T) {
 		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
 		assert.EventuallyWithT(t, func(t *assert.CollectT) {
 			organization, err := orgRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						orgRepo.IDCondition(organization.Id),
-						orgRepo.InstanceIDCondition(instanceID),
-					),
-				),
+				database.WithCondition(orgRepo.PrimaryKeyCondition(instanceID, organization.GetId())),
 			)
 			require.NoError(t, err)
 
@@ -234,12 +211,7 @@ func TestServer_TestOrganizationReduces(t *testing.T) {
 		retryDuration, tick := integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
 		assert.EventuallyWithT(t, func(t *assert.CollectT) {
 			_, err := orgRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						orgRepo.IDCondition(organization.Id),
-						orgRepo.InstanceIDCondition(instanceID),
-					),
-				),
+				database.WithCondition(orgRepo.PrimaryKeyCondition(instanceID, organization.GetId())),
 			)
 			require.NoError(t, err)
 		}, retryDuration, tick)
@@ -253,12 +225,7 @@ func TestServer_TestOrganizationReduces(t *testing.T) {
 		retryDuration, tick = integration.WaitForAndTickWithMaxDuration(CTX, time.Minute)
 		assert.EventuallyWithT(t, func(t *assert.CollectT) {
 			organization, err := orgRepo.Get(CTX, pool,
-				database.WithCondition(
-					database.And(
-						orgRepo.IDCondition(organization.Id),
-						orgRepo.InstanceIDCondition(instanceID),
-					),
-				),
+				database.WithCondition(orgRepo.PrimaryKeyCondition(instanceID, organization.GetId())),
 			)
 			require.ErrorIs(t, err, new(database.NoRowFoundError))
 
@@ -267,3 +234,37 @@ func TestServer_TestOrganizationReduces(t *testing.T) {
 		}, retryDuration, tick)
 	})
 }
+
+func createOrg(t *testing.T) *org.CreateOrganizationResponse {
+	t.Helper()
+	org, err := OrgClient.CreateOrganization(CTX, &org.CreateOrganizationRequest{
+		Name: gofakeit.Name(),
+	})
+	require.NoError(t, err)
+
+	orgRepo := repository.OrganizationRepository()
+	retryDuration, tick := integration.WaitForAndTickWithMaxDuration(t.Context(), time.Minute)
+	assert.EventuallyWithT(t, func(tc *assert.CollectT) {
+		_, err := orgRepo.Get(t.Context(), pool,
+			database.WithCondition(orgRepo.PrimaryKeyCondition(Instance.Instance.Id, org.GetId())),
+		)
+		assert.NoError(tc, err)
+	}, retryDuration, tick)
+
+	return org
+}
+
+func createTestScopedOrg(t *testing.T) *org.CreateOrganizationResponse {
+	org := createOrg(t)
+
+	t.Cleanup(func() {
+		_, err := OrgClient.DeleteOrganization(CTX, &v2beta.DeleteOrganizationRequest{
+			Id: org.GetId(),
+		})
+		if err != nil {
+			t.Logf("Failed to delete organization on cleanup: %v", err)
+		}
+	})
+
+	return org
+}

backend/v3/storage/database/query.go

@@ -3,6 +3,7 @@ package database
 import (
 	"fmt"
 	"reflect"
+	"slices"
 
 	"go.uber.org/mock/gomock"
 )
@@ -148,17 +149,15 @@ func (q *QueryOpts) Matches(x any) bool {
 	inputOpts.Write(inputBuilder)
 	q.Write(expectedBuilder)
 
-	if len(inputBuilder.Args()) != len(expectedBuilder.Args()) {
-		return false
+	deepEq := func(input, expected any) int {
+		if reflect.DeepEqual(input, expected) {
+			return 0
+		}
+		return -1
 	}
 
-	for i := range inputBuilder.Args() {
-		argIn, argExp := inputBuilder.Args()[i], expectedBuilder.Args()[i]
-
-		isEqual := reflect.DeepEqual(argIn, argExp)
-		if !isEqual {
-			return false
-		}
+	if slices.CompareFunc(inputBuilder.Args(), expectedBuilder.Args(), deepEq) != 0 {
+		return false
 	}
 	return inputBuilder.String() == expectedBuilder.String()
 }

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

@@ -247,12 +247,7 @@ func TestCreateOrganization(t *testing.T) {
 
 			// check organization values
 			organization, err = organizationRepo.Get(t.Context(), savepoint,
-				database.WithCondition(
-					database.And(
-						organizationRepo.IDCondition(organization.ID),
-						organizationRepo.InstanceIDCondition(organization.InstanceID),
-					),
-				),
+				database.WithCondition(organizationRepo.PrimaryKeyCondition(organization.InstanceID, organization.ID)),
 			)
 			require.NoError(t, err)
 
@@ -342,12 +337,7 @@ func TestUpdateOrganization(t *testing.T) {
 				require.NoError(t, err)
 
 				// delete instance
-				_, err = organizationRepo.Delete(t.Context(), tx,
-					database.And(
-						organizationRepo.InstanceIDCondition(org.InstanceID),
-						organizationRepo.IDCondition(org.ID),
-					),
-				)
+				_, err = organizationRepo.Delete(t.Context(), tx, organizationRepo.PrimaryKeyCondition(org.InstanceID, org.ID))
 				require.NoError(t, err)
 
 				return &org
@@ -399,10 +389,7 @@ func TestUpdateOrganization(t *testing.T) {
 
 			// update org
 			rowsAffected, err := organizationRepo.Update(t.Context(), tx,
-				database.And(
-					organizationRepo.InstanceIDCondition(createdOrg.InstanceID),
-					organizationRepo.IDCondition(createdOrg.ID),
-				),
+				organizationRepo.PrimaryKeyCondition(createdOrg.InstanceID, createdOrg.ID),
 				tt.update...,
 			)
 			afterUpdate := time.Now()
@@ -417,10 +404,7 @@ func TestUpdateOrganization(t *testing.T) {
 			// check organization values
 			organization, err := organizationRepo.Get(t.Context(), tx,
 				database.WithCondition(
-					database.And(
-						organizationRepo.IDCondition(createdOrg.ID),
-						organizationRepo.InstanceIDCondition(createdOrg.InstanceID),
-					),
+					organizationRepo.PrimaryKeyCondition(createdOrg.InstanceID, createdOrg.ID),
 				),
 			)
 			require.NoError(t, err)

backend/v3/storage/database/tx.go

@@ -4,10 +4,17 @@ import "context"
 
 // Transaction is an SQL transaction.
 type Transaction interface {
+	// Commit marks a transaction as successful and commits it to the database.
 	Commit(ctx context.Context) error
+	// Rollback undoes all changes made in the transaction.
 	Rollback(ctx context.Context) error
+	// End commits the transaction if err is nil, otherwise rollbacks.
+	//
+	// If err is nil the returned error is from Commit.
+	// If err is not nil, and rollback is successful, the original err is returned.
+	// If err is not nil, and rollback fails, the two errors are joined using [errors.Join].
 	End(ctx context.Context, err error) error
-
+	// Begin starts a sub transaction.
 	Begin(ctx context.Context) (Transaction, error)
 
 	QueryExecutor