perf(actionsv2): execution target router (#10564)

# Which Problems Are Solved

The event execution system currently uses a projection handler that
subscribes to and processes all events for all instances. This creates a
high static cost because the system over-fetches event data, handling
many events that are not needed by most instances. This inefficiency is
also reflected in high "rows returned" metrics in the database.

# How the Problems Are Solved

Eliminate the use of a project handler. Instead, events for which
"execution targets" are defined, are directly pushed to the queue by the
eventstore. A Router is populated in the Instance object in the authz
middleware.

- By joining the execution targets to the instance, no additional
queries are needed anymore.
- As part of the instance object, execution targets are now cached as
well.
- Events are queued within the same transaction, giving transactional
guarantees on delivery.
- Uses the "insert many fast` variant of River. Multiple jobs are queued
in a single round-trip to the database.
- Fix compatibility with PostgreSQL 15

# Additional Changes

- The signing key was stored as plain-text in the river job payload in
the DB. This violated our [Secrets
Storage](https://zitadel.com/docs/concepts/architecture/secrets#secrets-storage)
principle. This change removed the field and only uses the encrypted
version of the signing key.
- Fixed the target ordering from descending to ascending.
- Some minor linter warnings on the use of `io.WriteString()`.

# Additional Context

- Introduced in https://github.com/zitadel/zitadel/pull/9249
- Closes https://github.com/zitadel/zitadel/issues/10553
- Closes https://github.com/zitadel/zitadel/issues/9832
- Closes https://github.com/zitadel/zitadel/issues/10372
- Closes https://github.com/zitadel/zitadel/issues/10492

---------

Co-authored-by: Stefan Benz <46600784+stebenz@users.noreply.github.com>
(cherry picked from commit a9ebc06c77)

internal/eventstore/config.go

@@ -11,4 +11,5 @@ type Config struct {
 	Pusher   Pusher
 	Querier  Querier
 	Searcher Searcher
+	Queue    ExecutionQueue
 }

internal/eventstore/mock/queue.mock.go

@@ -0,0 +1,62 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: github.com/zitadel/zitadel/internal/eventstore (interfaces: ExecutionQueue)
+//
+// Generated by this command:
+//
+//	mockgen -package mock -destination ./mock/queue.mock.go github.com/zitadel/zitadel/internal/eventstore ExecutionQueue
+//
+
+// Package mock is a generated GoMock package.
+package mock
+
+import (
+	context "context"
+	sql "database/sql"
+	reflect "reflect"
+
+	river "github.com/riverqueue/river"
+	queue "github.com/zitadel/zitadel/internal/queue"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockExecutionQueue is a mock of ExecutionQueue interface.
+type MockExecutionQueue struct {
+	ctrl     *gomock.Controller
+	recorder *MockExecutionQueueMockRecorder
+}
+
+// MockExecutionQueueMockRecorder is the mock recorder for MockExecutionQueue.
+type MockExecutionQueueMockRecorder struct {
+	mock *MockExecutionQueue
+}
+
+// NewMockExecutionQueue creates a new mock instance.
+func NewMockExecutionQueue(ctrl *gomock.Controller) *MockExecutionQueue {
+	mock := &MockExecutionQueue{ctrl: ctrl}
+	mock.recorder = &MockExecutionQueueMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockExecutionQueue) EXPECT() *MockExecutionQueueMockRecorder {
+	return m.recorder
+}
+
+// InsertManyFastTx mocks base method.
+func (m *MockExecutionQueue) InsertManyFastTx(arg0 context.Context, arg1 *sql.Tx, arg2 []river.JobArgs, arg3 ...queue.InsertOpt) error {
+	m.ctrl.T.Helper()
+	varargs := []any{arg0, arg1, arg2}
+	for _, a := range arg3 {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "InsertManyFastTx", varargs...)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// InsertManyFastTx indicates an expected call of InsertManyFastTx.
+func (mr *MockExecutionQueueMockRecorder) InsertManyFastTx(arg0, arg1, arg2 any, arg3 ...any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]any{arg0, arg1, arg2}, arg3...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InsertManyFastTx", reflect.TypeOf((*MockExecutionQueue)(nil).InsertManyFastTx), varargs...)
+}

internal/eventstore/queue.go

@@ -0,0 +1,20 @@
+package eventstore
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/riverqueue/river"
+
+	"github.com/zitadel/zitadel/internal/queue"
+)
+
+//go:generate mockgen -package mock -destination ./mock/queue.mock.go github.com/zitadel/zitadel/internal/eventstore ExecutionQueue
+
+type ExecutionQueue interface {
+	// InsertManyFastTx wraps [river.Client.InsertManyFastTx] to insert all jobs in
+	// a single `COPY FROM` execution, within an existing transaction.
+	//
+	// Opts are applied to each job before sending them to river.
+	InsertManyFastTx(ctx context.Context, tx *sql.Tx, args []river.JobArgs, opts ...queue.InsertOpt) error
+}

internal/eventstore/v3/eventstore.go

@@ -33,6 +33,7 @@ var (
 
 type Eventstore struct {
 	client *database.DB
+	queue  eventstore.ExecutionQueue
 }
 
 var (
@@ -157,8 +158,14 @@ func (es *Eventstore) Client() *database.DB {
 	return es.client
 }
 
-func NewEventstore(client *database.DB) *Eventstore {
-	return &Eventstore{client: client}
+func NewEventstore(client *database.DB, opts ...EventstoreOption) *Eventstore {
+	es := &Eventstore{
+		client: client,
+	}
+	for _, opt := range opts {
+		opt(es)
+	}
+	return es
 }
 
 func (es *Eventstore) Health(ctx context.Context) error {
@@ -200,3 +207,11 @@ func (es *Eventstore) pushTx(ctx context.Context, client database.ContextQueryEx
 	}
 	return tx, func(err error) error { return database.CloseTransaction(tx, err) }, nil
 }
+
+type EventstoreOption func(*Eventstore)
+
+func WithExecutionQueueOption(queue eventstore.ExecutionQueue) EventstoreOption {
+	return func(es *Eventstore) {
+		es.queue = queue
+	}
+}

internal/eventstore/v3/mock_test.go

@@ -47,6 +47,10 @@ func (e *mockCommand) Fields() []*eventstore.FieldOperation {
 }
 
 func mockEvent(aggregate *eventstore.Aggregate, sequence uint64, payload Payload) eventstore.Event {
+	return mockEventType(aggregate, sequence, payload, "event.type")
+}
+
+func mockEventType(aggregate *eventstore.Aggregate, sequence uint64, payload Payload, typ string) eventstore.Event {
 	return &event{
 		command: &command{
 			InstanceID:    aggregate.InstanceID,
@@ -55,7 +59,7 @@ func mockEvent(aggregate *eventstore.Aggregate, sequence uint64, payload Payload
 			Owner:         aggregate.ResourceOwner,
 			Creator:       "creator",
 			Revision:      1,
-			CommandType:   "event.type",
+			CommandType:   typ,
 			Payload:       payload,
 		},
 		sequence: sequence,

internal/eventstore/v3/push.go

@@ -6,11 +6,14 @@ import (
 	_ "embed"
 	"fmt"
 
+	"github.com/riverqueue/river"
 	"github.com/zitadel/logging"
 
 	"github.com/zitadel/zitadel/internal/api/authz"
 	"github.com/zitadel/zitadel/internal/database"
 	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/queue"
+	exec_repo "github.com/zitadel/zitadel/internal/repository/execution"
 	"github.com/zitadel/zitadel/internal/telemetry/tracing"
 )
 
@@ -76,6 +79,11 @@ func (es *Eventstore) writeCommands(ctx context.Context, client database.Context
 		return nil, err
 	}
 
+	err = es.queueExecutions(ctx, tx, events)
+	if err != nil {
+		return nil, err
+	}
+
 	return events, nil
 }
 
@@ -106,3 +114,54 @@ func writeEvents(ctx context.Context, tx database.Tx, commands []eventstore.Comm
 	}
 	return events, nil
 }
+
+func (es *Eventstore) queueExecutions(ctx context.Context, tx database.Tx, events []eventstore.Event) error {
+	if es.queue == nil {
+		return nil
+	}
+
+	sqlTx, ok := tx.(*sql.Tx)
+	if !ok {
+		types := make([]string, len(events))
+		for i, event := range events {
+			types[i] = string(event.Type())
+		}
+		logging.WithFields("event_types", types).Warningf("event executions skipped: wrong type of transaction %T", tx)
+		return nil
+	}
+	jobArgs, err := eventsToJobArgs(ctx, events)
+	if err != nil {
+		return err
+	}
+	if len(jobArgs) == 0 {
+		return nil
+	}
+	return es.queue.InsertManyFastTx(
+		ctx, sqlTx, jobArgs,
+		queue.WithQueueName(exec_repo.QueueName),
+	)
+}
+
+func eventsToJobArgs(ctx context.Context, events []eventstore.Event) ([]river.JobArgs, error) {
+	if len(events) == 0 {
+		return nil, nil
+	}
+	router := authz.GetInstance(ctx).ExecutionRouter()
+	if router.IsZero() {
+		return nil, nil
+	}
+
+	jobArgs := make([]river.JobArgs, 0, len(events))
+	for _, event := range events {
+		targets, ok := router.GetEventBestMatch(fmt.Sprintf("event/%s", event.Type()))
+		if !ok {
+			continue
+		}
+		req, err := exec_repo.NewRequest(event, targets)
+		if err != nil {
+			return nil, err
+		}
+		jobArgs = append(jobArgs, req)
+	}
+	return jobArgs, nil
+}

internal/eventstore/v3/push_test.go

@@ -1,15 +1,23 @@
 package eventstore
 
 import (
+	"context"
+	"database/sql"
 	_ "embed"
 	"testing"
 
+	"github.com/riverqueue/river"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 
+	"github.com/zitadel/zitadel/internal/api/authz"
 	"github.com/zitadel/zitadel/internal/database"
 	"github.com/zitadel/zitadel/internal/database/postgres"
 	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/eventstore/mock"
+	"github.com/zitadel/zitadel/internal/execution/target"
+	exec_repo "github.com/zitadel/zitadel/internal/repository/execution"
 )
 
 func Test_mapCommands(t *testing.T) {
@@ -251,3 +259,159 @@ func Test_mapCommands(t *testing.T) {
 		})
 	}
 }
+
+func TestEventstore_queueExecutions(t *testing.T) {
+	events := []eventstore.Event{
+		mockEventType(mockAggregate("TEST"), 1, nil, "ex.foo.bar"),
+		mockEventType(mockAggregate("TEST"), 2, nil, "ex.bar.foo"),
+		mockEventType(mockAggregate("TEST"), 3, nil, "ex.removed"),
+	}
+	type args struct {
+		ctx    context.Context
+		tx     database.Tx
+		events []eventstore.Event
+	}
+	tests := []struct {
+		name    string
+		queue   func(t *testing.T) eventstore.ExecutionQueue
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "incorrect Tx type, noop",
+			queue: func(t *testing.T) eventstore.ExecutionQueue {
+				mQueue := mock.NewMockExecutionQueue(gomock.NewController(t))
+				return mQueue
+			},
+			args: args{
+				ctx:    context.Background(),
+				tx:     nil,
+				events: events,
+			},
+			wantErr: false,
+		},
+		{
+			name: "no events",
+			queue: func(t *testing.T) eventstore.ExecutionQueue {
+				mQueue := mock.NewMockExecutionQueue(gomock.NewController(t))
+				return mQueue
+			},
+			args: args{
+				ctx:    context.Background(),
+				tx:     &sql.Tx{},
+				events: []eventstore.Event{},
+			},
+			wantErr: false,
+		},
+		{
+			name: "no router in Ctx",
+			queue: func(t *testing.T) eventstore.ExecutionQueue {
+				mQueue := mock.NewMockExecutionQueue(gomock.NewController(t))
+				return mQueue
+			},
+			args: args{
+				ctx:    context.Background(),
+				tx:     &sql.Tx{},
+				events: events,
+			},
+			wantErr: false,
+		},
+		{
+			name: "not found in router",
+			queue: func(t *testing.T) eventstore.ExecutionQueue {
+				mQueue := mock.NewMockExecutionQueue(gomock.NewController(t))
+				return mQueue
+			},
+			args: args{
+				ctx: authz.WithExecutionRouter(
+					context.Background(),
+					target.NewRouter([]target.Target{
+						{
+							ExecutionID: "function/fooBar",
+						},
+					}),
+				),
+				tx:     &sql.Tx{},
+				events: events,
+			},
+			wantErr: false,
+		},
+		{
+			name: "event prefix",
+			queue: func(t *testing.T) eventstore.ExecutionQueue {
+				mQueue := mock.NewMockExecutionQueue(gomock.NewController(t))
+				mQueue.EXPECT().InsertManyFastTx(
+					gomock.Any(),
+					gomock.Any(),
+					[]river.JobArgs{
+						mustNewRequest(t, events[0], []target.Target{{ExecutionID: "event"}}),
+						mustNewRequest(t, events[1], []target.Target{{ExecutionID: "event"}}),
+						mustNewRequest(t, events[2], []target.Target{{ExecutionID: "event"}}),
+					},
+					gomock.Any(),
+				)
+				return mQueue
+			},
+			args: args{
+				ctx: authz.WithExecutionRouter(
+					context.Background(),
+					target.NewRouter([]target.Target{
+						{ExecutionID: "function/fooBar"},
+						{ExecutionID: "event"},
+					}),
+				),
+				tx:     &sql.Tx{},
+				events: events,
+			},
+			wantErr: false,
+		},
+		{
+			name: "event wildcard and exact match",
+			queue: func(t *testing.T) eventstore.ExecutionQueue {
+				mQueue := mock.NewMockExecutionQueue(gomock.NewController(t))
+				mQueue.EXPECT().InsertManyFastTx(
+					gomock.Any(),
+					gomock.Any(),
+					[]river.JobArgs{
+						mustNewRequest(t, events[0], []target.Target{{ExecutionID: "event/ex.foo.*"}}),
+						mustNewRequest(t, events[2], []target.Target{{ExecutionID: "event/ex.removed"}}),
+					},
+					gomock.Any(),
+				)
+				return mQueue
+			},
+			args: args{
+				ctx: authz.WithExecutionRouter(
+					context.Background(),
+					target.NewRouter([]target.Target{
+						{ExecutionID: "function/fooBar"},
+						{ExecutionID: "event/ex.foo.*"},
+						{ExecutionID: "event/ex.removed"},
+					}),
+				),
+				tx:     &sql.Tx{},
+				events: events,
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			es := &Eventstore{
+				queue: tt.queue(t),
+			}
+			err := es.queueExecutions(tt.args.ctx, tt.args.tx, tt.args.events)
+			if tt.wantErr {
+				assert.Error(t, err)
+				return
+			}
+			assert.NoError(t, err)
+		})
+	}
+}
+
+func mustNewRequest(t *testing.T, e eventstore.Event, targets []target.Target) *exec_repo.Request {
+	req, err := exec_repo.NewRequest(e, targets)
+	require.NoError(t, err, "exec_repo.NewRequest")
+	return req
+}