feat: add action v2 execution on requests and responses (#7637)

* feat: add execution of targets to grpc calls * feat: add execution of targets to grpc calls * feat: add execution of targets to grpc calls * feat: add execution of targets to grpc calls * feat: add execution of targets to grpc calls * feat: add execution of targets to grpc calls * feat: add execution of targets to grpc calls * feat: split request and response logic to handle the different context information * feat: split request and response logic to handle the different context information * fix: integration test * fix: import alias * fix: refactor execution package * fix: refactor execution interceptor integration and unit tests * fix: refactor execution interceptor integration and unit tests * fix: refactor execution interceptor integration and unit tests * fix: refactor execution interceptor integration and unit tests * fix: refactor execution interceptor integration and unit tests * docs: basic documentation for executions and targets * fix: change order for interceptors * fix: merge back origin/main * fix: change target definition command and query side (#7735) * fix: change target definition command and query side * fix: correct refactoring name changes * fix: correct refactoring name changes * fix: changing execution defintion with target list and type * fix: changing execution definition with target list and type * fix: add back search queries for target and include * fix: projections change for execution with targets suffix table * fix: projections change for execution with targets suffix table * fix: projections change for execution with targets suffix table * fix: projections change for execution with targets suffix table * fix: projections change for execution with targets suffix table * fix: projections change for execution with targets suffix table * fix: projections change for execution with targets suffix table * docs: add example to actions v2 * docs: add example to actions v2 * fix: correct integration tests on query for executions * fix: add separate event for execution v2 as content changed * fix: add separate event for execution v2 as content changed * fix: added review comment changes * fix: added review comment changes * fix: added review comment changes --------- Co-authored-by: adlerhurst <silvan.reusser@gmail.com> * fix: added review comment changes * fix: added review comment changes * Update internal/api/grpc/server/middleware/execution_interceptor.go Co-authored-by: Silvan <silvan.reusser@gmail.com> * fix: added review comment changes * fix: added review comment changes * fix: added review comment changes * fix: added review comment changes * fix: added review comment changes * fix: added review comment changes --------- Co-authored-by: adlerhurst <silvan.reusser@gmail.com> Co-authored-by: Elio Bischof <elio@zitadel.com>
2025-08-12 00:27:31 +00:00 · 2024-05-04 11:55:57 +02:00
parent 7e345444bf
commit 1c5ecba42a
67 changed files with 4397 additions and 1556 deletions
--- a/internal/query/projection/execution.go
+++ b/internal/query/projection/execution.go
@@ -3,6 +3,7 @@ package projection
 import (
 	"context"

+	"github.com/zitadel/zitadel/internal/domain"
 	"github.com/zitadel/zitadel/internal/eventstore"
 	old_handler "github.com/zitadel/zitadel/internal/eventstore/handler"
 	"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
@@ -11,15 +12,19 @@ import (
 )

 const (
-	ExecutionTable            = "projections.executions"
-	ExecutionIDCol            = "id"
-	ExecutionCreationDateCol  = "creation_date"
-	ExecutionChangeDateCol    = "change_date"
-	ExecutionResourceOwnerCol = "resource_owner"
-	ExecutionInstanceIDCol    = "instance_id"
-	ExecutionSequenceCol      = "sequence"
-	ExecutionTargetsCol       = "targets"
-	ExecutionIncludesCol      = "includes"
+	ExecutionTable           = "projections.executions1"
+	ExecutionIDCol           = "id"
+	ExecutionCreationDateCol = "creation_date"
+	ExecutionChangeDateCol   = "change_date"
+	ExecutionInstanceIDCol   = "instance_id"
+	ExecutionSequenceCol     = "sequence"
+
+	ExecutionTargetSuffix         = "targets"
+	ExecutionTargetExecutionIDCol = "execution_id"
+	ExecutionTargetInstanceIDCol  = "instance_id"
+	ExecutionTargetPositionCol    = "position"
+	ExecutionTargetTargetIDCol    = "target_id"
+	ExecutionTargetIncludeCol     = "include"
 )

 type executionProjection struct{}
@@ -33,19 +38,28 @@ func (*executionProjection) Name() string {
 }

 func (*executionProjection) Init() *old_handler.Check {
-	return handler.NewTableCheck(
+	return handler.NewMultiTableCheck(
 		handler.NewTable([]*handler.InitColumn{
 			handler.NewColumn(ExecutionIDCol, handler.ColumnTypeText),
 			handler.NewColumn(ExecutionCreationDateCol, handler.ColumnTypeTimestamp),
 			handler.NewColumn(ExecutionChangeDateCol, handler.ColumnTypeTimestamp),
-			handler.NewColumn(ExecutionResourceOwnerCol, handler.ColumnTypeText),
-			handler.NewColumn(ExecutionInstanceIDCol, handler.ColumnTypeText),
 			handler.NewColumn(ExecutionSequenceCol, handler.ColumnTypeInt64),
-			handler.NewColumn(ExecutionTargetsCol, handler.ColumnTypeTextArray, handler.Nullable()),
-			handler.NewColumn(ExecutionIncludesCol, handler.ColumnTypeTextArray, handler.Nullable()),
+			handler.NewColumn(ExecutionInstanceIDCol, handler.ColumnTypeText),
 		},
 			handler.NewPrimaryKey(ExecutionInstanceIDCol, ExecutionIDCol),
 		),
+		handler.NewSuffixedTable([]*handler.InitColumn{
+			handler.NewColumn(ExecutionTargetInstanceIDCol, handler.ColumnTypeText),
+			handler.NewColumn(ExecutionTargetExecutionIDCol, handler.ColumnTypeText),
+			handler.NewColumn(ExecutionTargetPositionCol, handler.ColumnTypeInt64),
+			handler.NewColumn(ExecutionTargetIncludeCol, handler.ColumnTypeText, handler.Nullable()),
+			handler.NewColumn(ExecutionTargetTargetIDCol, handler.ColumnTypeText, handler.Nullable()),
+		},
+			handler.NewPrimaryKey(ExecutionTargetInstanceIDCol, ExecutionTargetExecutionIDCol, ExecutionTargetPositionCol),
+			ExecutionTargetSuffix,
+			handler.WithForeignKey(handler.NewForeignKey("execution", []string{ExecutionTargetInstanceIDCol, ExecutionTargetExecutionIDCol}, []string{ExecutionInstanceIDCol, ExecutionIDCol})),
+			handler.WithIndex(handler.NewIndex("execution", []string{ExecutionTargetInstanceIDCol, ExecutionTargetExecutionIDCol})),
+		),
 	)
 }

@@ -55,7 +69,7 @@ func (p *executionProjection) Reducers() []handler.AggregateReducer {
 			Aggregate: exec.AggregateType,
 			EventReducers: []handler.EventReducer{
 				{
-					Event:  exec.SetEventType,
+					Event:  exec.SetEventV2Type,
 					Reduce: p.reduceExecutionSet,
 				},
 				{
@@ -77,21 +91,65 @@ func (p *executionProjection) Reducers() []handler.AggregateReducer {
 }

 func (p *executionProjection) reduceExecutionSet(event eventstore.Event) (*handler.Statement, error) {
-	e, err := assertEvent[*exec.SetEvent](event)
+	e, err := assertEvent[*exec.SetEventV2](event)
 	if err != nil {
 		return nil, err
 	}
-	columns := []handler.Column{
-		handler.NewCol(ExecutionInstanceIDCol, e.Aggregate().InstanceID),
-		handler.NewCol(ExecutionIDCol, e.Aggregate().ID),
-		handler.NewCol(ExecutionResourceOwnerCol, e.Aggregate().ResourceOwner),
-		handler.NewCol(ExecutionCreationDateCol, handler.OnlySetValueOnInsert(ExecutionTable, e.CreationDate())),
-		handler.NewCol(ExecutionChangeDateCol, e.CreationDate()),
-		handler.NewCol(ExecutionSequenceCol, e.Sequence()),
-		handler.NewCol(ExecutionTargetsCol, e.Targets),
-		handler.NewCol(ExecutionIncludesCol, e.Includes),
+
+	stmts := []func(eventstore.Event) handler.Exec{
+		handler.AddUpsertStatement(
+			[]handler.Column{
+				handler.NewCol(ExecutionInstanceIDCol, e.Aggregate().InstanceID),
+				handler.NewCol(ExecutionIDCol, e.Aggregate().ID),
+			},
+			[]handler.Column{
+				handler.NewCol(ExecutionInstanceIDCol, e.Aggregate().InstanceID),
+				handler.NewCol(ExecutionIDCol, e.Aggregate().ID),
+				handler.NewCol(ExecutionCreationDateCol, handler.OnlySetValueOnInsert(ExecutionTable, e.CreationDate())),
+				handler.NewCol(ExecutionChangeDateCol, e.CreationDate()),
+				handler.NewCol(ExecutionSequenceCol, e.Sequence()),
+			},
+		),
+		// cleanup execution targets to re-insert them
+		handler.AddDeleteStatement(
+			[]handler.Condition{
+				handler.NewCond(ExecutionTargetInstanceIDCol, e.Aggregate().InstanceID),
+				handler.NewCond(ExecutionTargetExecutionIDCol, e.Aggregate().ID),
+			},
+			handler.WithTableSuffix(ExecutionTargetSuffix),
+		),
 	}
-	return handler.NewUpsertStatement(e, columns[0:2], columns), nil
+
+	if len(e.Targets) > 0 {
+		for i, target := range e.Targets {
+			var targetStr, includeStr string
+			switch target.Type {
+			case domain.ExecutionTargetTypeTarget:
+				targetStr = target.Target
+			case domain.ExecutionTargetTypeInclude:
+				includeStr = target.Target
+			case domain.ExecutionTargetTypeUnspecified:
+				continue
+			default:
+				continue
+			}
+
+			stmts = append(stmts,
+				handler.AddCreateStatement(
+					[]handler.Column{
+						handler.NewCol(ExecutionTargetInstanceIDCol, e.Aggregate().InstanceID),
+						handler.NewCol(ExecutionTargetExecutionIDCol, e.Aggregate().ID),
+						handler.NewCol(ExecutionTargetPositionCol, i+1),
+						handler.NewCol(ExecutionTargetIncludeCol, includeStr),
+						handler.NewCol(ExecutionTargetTargetIDCol, targetStr),
+					},
+					handler.WithTableSuffix(ExecutionTargetSuffix),
+				),
+			)
+		}
+	}
+
+	return handler.NewMultiStatement(e, stmts...), nil
 }

 func (p *executionProjection) reduceExecutionRemoved(event eventstore.Event) (*handler.Statement, error) {
@@ -99,8 +157,8 @@ func (p *executionProjection) reduceExecutionRemoved(event eventstore.Event) (*h
 	if err != nil {
 		return nil, err
 	}
-	return handler.NewDeleteStatement(
-		e,
+
+	return handler.NewDeleteStatement(e,
 		[]handler.Condition{
 			handler.NewCond(ExecutionInstanceIDCol, e.Aggregate().InstanceID),
 			handler.NewCond(ExecutionIDCol, e.Aggregate().ID),
--- a/internal/query/projection/execution_test.go
+++ b/internal/query/projection/execution_test.go
@@ -25,11 +25,11 @@ func TestExecutionProjection_reduces(t *testing.T) {
 			args: args{
 				event: getEvent(
 					testEvent(
-						exec.SetEventType,
+						exec.SetEventV2Type,
 						exec.AggregateType,
-						[]byte(`{"targets": ["target"], "includes": ["include"]}`),
+						[]byte(`{"targets": [{"type":2,"target":"target"},{"type":1,"target":"include"}]}`),
 					),
-					eventstore.GenericEventMapper[exec.SetEvent],
+					eventstore.GenericEventMapper[exec.SetEventV2],
 				),
 			},
 			reduce: (&executionProjection{}).reduceExecutionSet,
@@ -39,16 +39,40 @@ func TestExecutionProjection_reduces(t *testing.T) {
 				executer: &testExecuter{
 					executions: []execution{
 						{
-							expectedStmt: "INSERT INTO projections.executions (instance_id, id, resource_owner, creation_date, change_date, sequence, targets, includes) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (instance_id, id) DO UPDATE SET (resource_owner, creation_date, change_date, sequence, targets, includes) = (EXCLUDED.resource_owner, projections.executions.creation_date, EXCLUDED.change_date, EXCLUDED.sequence, EXCLUDED.targets, EXCLUDED.includes)",
+							expectedStmt: "INSERT INTO projections.executions1 (instance_id, id, creation_date, change_date, sequence) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (instance_id, id) DO UPDATE SET (creation_date, change_date, sequence) = (projections.executions1.creation_date, EXCLUDED.change_date, EXCLUDED.sequence)",
 							expectedArgs: []interface{}{
 								"instance-id",
 								"agg-id",
-								"ro-id",
 								anyArg{},
 								anyArg{},
 								uint64(15),
-								[]string{"target"},
-								[]string{"include"},
+							},
+						},
+						{
+							expectedStmt: "DELETE FROM projections.executions1_targets WHERE (instance_id = $1) AND (execution_id = $2)",
+							expectedArgs: []interface{}{
+								"instance-id",
+								"agg-id",
+							},
+						},
+						{
+							expectedStmt: "INSERT INTO projections.executions1_targets (instance_id, execution_id, position, include, target_id) VALUES ($1, $2, $3, $4, $5)",
+							expectedArgs: []interface{}{
+								"instance-id",
+								"agg-id",
+								1,
+								"",
+								"target",
+							},
+						},
+						{
+							expectedStmt: "INSERT INTO projections.executions1_targets (instance_id, execution_id, position, include, target_id) VALUES ($1, $2, $3, $4, $5)",
+							expectedArgs: []interface{}{
+								"instance-id",
+								"agg-id",
+								2,
+								"include",
+								"",
 							},
 						},
 					},
@@ -74,7 +98,7 @@ func TestExecutionProjection_reduces(t *testing.T) {
 				executer: &testExecuter{
 					executions: []execution{
 						{
-							expectedStmt: "DELETE FROM projections.executions WHERE (instance_id = $1) AND (id = $2)",
+							expectedStmt: "DELETE FROM projections.executions1 WHERE (instance_id = $1) AND (id = $2)",
 							expectedArgs: []interface{}{
 								"instance-id",
 								"agg-id",
@@ -103,7 +127,7 @@ func TestExecutionProjection_reduces(t *testing.T) {
 				executer: &testExecuter{
 					executions: []execution{
 						{
-							expectedStmt: "DELETE FROM projections.executions WHERE (instance_id = $1)",
+							expectedStmt: "DELETE FROM projections.executions1 WHERE (instance_id = $1)",
 							expectedArgs: []interface{}{
 								"agg-id",
 							},
--- a/internal/query/projection/projection.go
+++ b/internal/query/projection/projection.go
@@ -269,8 +269,8 @@ func newProjectionsList() {
 		RestrictionsProjection,
 		SystemFeatureProjection,
 		InstanceFeatureProjection,
-		ExecutionProjection,
 		TargetProjection,
+		ExecutionProjection,
 		UserSchemaProjection,
 	}
 }
--- a/internal/query/projection/target.go
+++ b/internal/query/projection/target.go
@@ -11,7 +11,7 @@ import (
 )

 const (
-	TargetTable               = "projections.targets"
+	TargetTable               = "projections.targets1"
 	TargetIDCol               = "id"
 	TargetCreationDateCol     = "creation_date"
 	TargetChangeDateCol       = "change_date"
@@ -20,9 +20,8 @@ const (
 	TargetSequenceCol         = "sequence"
 	TargetNameCol             = "name"
 	TargetTargetType          = "target_type"
-	TargetURLCol              = "url"
+	TargetEndpointCol         = "endpoint"
 	TargetTimeoutCol          = "timeout"
-	TargetAsyncCol            = "async"
 	TargetInterruptOnErrorCol = "interrupt_on_error"
 )

@@ -47,10 +46,9 @@ func (*targetProjection) Init() *old_handler.Check {
 			handler.NewColumn(TargetTargetType, handler.ColumnTypeEnum),
 			handler.NewColumn(TargetSequenceCol, handler.ColumnTypeInt64),
 			handler.NewColumn(TargetNameCol, handler.ColumnTypeText),
-			handler.NewColumn(TargetURLCol, handler.ColumnTypeText, handler.Default("")),
-			handler.NewColumn(TargetTimeoutCol, handler.ColumnTypeInt64, handler.Default(0)),
-			handler.NewColumn(TargetAsyncCol, handler.ColumnTypeBool, handler.Default(false)),
-			handler.NewColumn(TargetInterruptOnErrorCol, handler.ColumnTypeBool, handler.Default(false)),
+			handler.NewColumn(TargetEndpointCol, handler.ColumnTypeText),
+			handler.NewColumn(TargetTimeoutCol, handler.ColumnTypeInt64),
+			handler.NewColumn(TargetInterruptOnErrorCol, handler.ColumnTypeBool),
 		},
 			handler.NewPrimaryKey(TargetInstanceIDCol, TargetIDCol),
 		),
@@ -103,10 +101,9 @@ func (p *targetProjection) reduceTargetAdded(event eventstore.Event) (*handler.S
 			handler.NewCol(TargetChangeDateCol, e.CreationDate()),
 			handler.NewCol(TargetSequenceCol, e.Sequence()),
 			handler.NewCol(TargetNameCol, e.Name),
-			handler.NewCol(TargetURLCol, e.URL),
+			handler.NewCol(TargetEndpointCol, e.Endpoint),
 			handler.NewCol(TargetTargetType, e.TargetType),
 			handler.NewCol(TargetTimeoutCol, e.Timeout),
-			handler.NewCol(TargetAsyncCol, e.Async),
 			handler.NewCol(TargetInterruptOnErrorCol, e.InterruptOnError),
 		},
 	), nil
@@ -128,15 +125,12 @@ func (p *targetProjection) reduceTargetChanged(event eventstore.Event) (*handler
 	if e.TargetType != nil {
 		values = append(values, handler.NewCol(TargetTargetType, *e.TargetType))
 	}
-	if e.URL != nil {
-		values = append(values, handler.NewCol(TargetURLCol, *e.URL))
+	if e.Endpoint != nil {
+		values = append(values, handler.NewCol(TargetEndpointCol, *e.Endpoint))
 	}
 	if e.Timeout != nil {
 		values = append(values, handler.NewCol(TargetTimeoutCol, *e.Timeout))
 	}
-	if e.Async != nil {
-		values = append(values, handler.NewCol(TargetAsyncCol, *e.Async))
-	}
 	if e.InterruptOnError != nil {
 		values = append(values, handler.NewCol(TargetInterruptOnErrorCol, *e.InterruptOnError))
 	}
--- a/internal/query/projection/target_test.go
+++ b/internal/query/projection/target_test.go
@@ -29,7 +29,7 @@ func TestTargetProjection_reduces(t *testing.T) {
 					testEvent(
 						target.AddedEventType,
 						target.AggregateType,
-						[]byte(`{"name": "name", "targetType":0, "url":"https://example.com", "timeout": 3000000000, "async": true, "interruptOnError": true}`),
+						[]byte(`{"name": "name", "targetType":0, "endpoint":"https://example.com", "timeout": 3000000000, "async": true, "interruptOnError": true}`),
 					),
 					eventstore.GenericEventMapper[target.AddedEvent],
 				),
@@ -41,7 +41,7 @@ func TestTargetProjection_reduces(t *testing.T) {
 				executer: &testExecuter{
 					executions: []execution{
 						{
-							expectedStmt: "INSERT INTO projections.targets (instance_id, resource_owner, id, creation_date, change_date, sequence, name, url, target_type, timeout, async, interrupt_on_error) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)",
+							expectedStmt: "INSERT INTO projections.targets1 (instance_id, resource_owner, id, creation_date, change_date, sequence, name, endpoint, target_type, timeout, interrupt_on_error) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)",
 							expectedArgs: []interface{}{
 								"instance-id",
 								"ro-id",
@@ -54,7 +54,6 @@ func TestTargetProjection_reduces(t *testing.T) {
 								domain.TargetTypeWebhook,
 								3 * time.Second,
 								true,
-								true,
 							},
 						},
 					},
@@ -68,7 +67,7 @@ func TestTargetProjection_reduces(t *testing.T) {
 					testEvent(
 						target.ChangedEventType,
 						target.AggregateType,
-						[]byte(`{"name": "name2", "targetType":0, "url":"https://example.com", "timeout": 3000000000, "async": true, "interruptOnError": true}`),
+						[]byte(`{"name": "name2", "targetType":0, "endpoint":"https://example.com", "timeout": 3000000000, "async": true, "interruptOnError": true}`),
 					),
 					eventstore.GenericEventMapper[target.ChangedEvent],
 				),
@@ -80,7 +79,7 @@ func TestTargetProjection_reduces(t *testing.T) {
 				executer: &testExecuter{
 					executions: []execution{
 						{
-							expectedStmt: "UPDATE projections.targets SET (change_date, sequence, resource_owner, name, target_type, url, timeout, async, interrupt_on_error) = ($1, $2, $3, $4, $5, $6, $7, $8, $9) WHERE (instance_id = $10) AND (id = $11)",
+							expectedStmt: "UPDATE projections.targets1 SET (change_date, sequence, resource_owner, name, target_type, endpoint, timeout, interrupt_on_error) = ($1, $2, $3, $4, $5, $6, $7, $8) WHERE (instance_id = $9) AND (id = $10)",
 							expectedArgs: []interface{}{
 								anyArg{},
 								uint64(15),
@@ -90,7 +89,6 @@ func TestTargetProjection_reduces(t *testing.T) {
 								"https://example.com",
 								3 * time.Second,
 								true,
-								true,
 								"instance-id",
 								"agg-id",
 							},
@@ -118,7 +116,7 @@ func TestTargetProjection_reduces(t *testing.T) {
 				executer: &testExecuter{
 					executions: []execution{
 						{
-							expectedStmt: "DELETE FROM projections.targets WHERE (instance_id = $1) AND (id = $2)",
+							expectedStmt: "DELETE FROM projections.targets1 WHERE (instance_id = $1) AND (id = $2)",
 							expectedArgs: []interface{}{
 								"instance-id",
 								"agg-id",
@@ -147,7 +145,7 @@ func TestTargetProjection_reduces(t *testing.T) {
 				executer: &testExecuter{
 					executions: []execution{
 						{
-							expectedStmt: "DELETE FROM projections.targets WHERE (instance_id = $1)",
+							expectedStmt: "DELETE FROM projections.targets1 WHERE (instance_id = $1)",
 							expectedArgs: []interface{}{
 								"agg-id",
 							},