mirror of
https://github.com/zitadel/zitadel.git
synced 2025-08-12 11:27:33 +00:00
feat: query side for executions and targets for actions v2 (#7524)
* feat: add projections and query side to executions and targets * feat: add list and get endpoints for targets * feat: add integration tests for query endpoints target and execution * fix: linting * fix: linting * fix: review changes, renames and corrections * fix: review changes, renames and corrections * fix: review changes, renames and corrections * fix: review changes, renames and corrections * fix: review changes, renames and corrections * fix: review changes, renames and corrections * fix: remove position from list details
This commit is contained in:
Stefan Benz
@@ -12,7 +12,6 @@ import (
|
||||
sq "github.com/Masterminds/squirrel"
|
||||
"github.com/zitadel/logging"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/api/authz"
|
||||
"github.com/zitadel/zitadel/internal/api/call"
|
||||
"github.com/zitadel/zitadel/internal/eventstore"
|
||||
"github.com/zitadel/zitadel/internal/query/projection"
|
||||
@@ -20,6 +19,10 @@ import (
|
||||
"github.com/zitadel/zitadel/internal/zerrors"
|
||||
)
|
||||
|
||||
type Stateful interface {
|
||||
SetState(*State)
|
||||
}
|
||||
|
||||
type State struct {
|
||||
LastRun time.Time
|
||||
|
||||
@@ -83,29 +86,7 @@ func (q *Queries) SearchCurrentStates(ctx context.Context, queries *CurrentState
|
||||
}
|
||||
|
||||
func (q *Queries) latestState(ctx context.Context, projections ...table) (state *State, err error) {
|
||||
ctx, span := tracing.NewSpan(ctx)
|
||||
defer func() { span.EndWithError(err) }()
|
||||
|
||||
query, scan := prepareLatestState(ctx, q.client)
|
||||
or := make(sq.Or, len(projections))
|
||||
for i, projection := range projections {
|
||||
or[i] = sq.Eq{CurrentStateColProjectionName.identifier(): projection.name}
|
||||
}
|
||||
stmt, args, err := query.
|
||||
Where(or).
|
||||
Where(sq.Eq{CurrentStateColInstanceID.identifier(): authz.GetInstance(ctx).InstanceID()}).
|
||||
OrderBy(CurrentStateColEventDate.identifier() + " DESC").
|
||||
ToSql()
|
||||
if err != nil {
|
||||
return nil, zerrors.ThrowInternal(err, "QUERY-5CfX9", "Errors.Query.SQLStatement")
|
||||
}
|
||||
|
||||
err = q.client.QueryRowContext(ctx, func(row *sql.Row) error {
|
||||
state, err = scan(row)
|
||||
return err
|
||||
}, stmt, args...)
|
||||
|
||||
return state, err
|
||||
return latestState(ctx, q.client, projections...)
|
||||
}
|
||||
|
||||
func (q *Queries) ClearCurrentSequence(ctx context.Context, projectionName string) (err error) {
|
||||
|
||||
191
internal/query/execution.go
Normal file
191
internal/query/execution.go
Normal file
@@ -0,0 +1,191 @@
|
||||
package query
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
|
||||
sq "github.com/Masterminds/squirrel"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/api/authz"
|
||||
"github.com/zitadel/zitadel/internal/database"
|
||||
"github.com/zitadel/zitadel/internal/domain"
|
||||
"github.com/zitadel/zitadel/internal/query/projection"
|
||||
"github.com/zitadel/zitadel/internal/zerrors"
|
||||
)
|
||||
|
||||
var (
|
||||
executionTable = table{
|
||||
name: projection.ExecutionTable,
|
||||
instanceIDCol: projection.ExecutionInstanceIDCol,
|
||||
}
|
||||
ExecutionColumnID = Column{
|
||||
name: projection.ExecutionIDCol,
|
||||
table: executionTable,
|
||||
}
|
||||
ExecutionColumnCreationDate = Column{
|
||||
name: projection.ExecutionCreationDateCol,
|
||||
table: executionTable,
|
||||
}
|
||||
ExecutionColumnChangeDate = Column{
|
||||
name: projection.ExecutionChangeDateCol,
|
||||
table: executionTable,
|
||||
}
|
||||
ExecutionColumnResourceOwner = Column{
|
||||
name: projection.ExecutionResourceOwnerCol,
|
||||
table: executionTable,
|
||||
}
|
||||
ExecutionColumnInstanceID = Column{
|
||||
name: projection.ExecutionInstanceIDCol,
|
||||
table: executionTable,
|
||||
}
|
||||
ExecutionColumnSequence = Column{
|
||||
name: projection.ExecutionSequenceCol,
|
||||
table: executionTable,
|
||||
}
|
||||
ExecutionColumnTargets = Column{
|
||||
name: projection.ExecutionTargetsCol,
|
||||
table: executionTable,
|
||||
}
|
||||
ExecutionColumnIncludes = Column{
|
||||
name: projection.ExecutionIncludesCol,
|
||||
table: executionTable,
|
||||
}
|
||||
)
|
||||
|
||||
type Executions struct {
|
||||
SearchResponse
|
||||
Executions []*Execution
|
||||
}
|
||||
|
||||
func (e *Executions) SetState(s *State) {
|
||||
e.State = s
|
||||
}
|
||||
|
||||
type Execution struct {
|
||||
ID string
|
||||
domain.ObjectDetails
|
||||
|
||||
Targets database.TextArray[string]
|
||||
Includes database.TextArray[string]
|
||||
}
|
||||
|
||||
type ExecutionSearchQueries struct {
|
||||
SearchRequest
|
||||
Queries []SearchQuery
|
||||
}
|
||||
|
||||
func (q *ExecutionSearchQueries) toQuery(query sq.SelectBuilder) sq.SelectBuilder {
|
||||
query = q.SearchRequest.toQuery(query)
|
||||
for _, q := range q.Queries {
|
||||
query = q.toQuery(query)
|
||||
}
|
||||
return query
|
||||
}
|
||||
|
||||
func (q *Queries) SearchExecutions(ctx context.Context, queries *ExecutionSearchQueries) (executions *Executions, err error) {
|
||||
eq := sq.Eq{
|
||||
ExecutionColumnInstanceID.identifier(): authz.GetInstance(ctx).InstanceID(),
|
||||
}
|
||||
query, scan := prepareExecutionsQuery(ctx, q.client)
|
||||
return genericRowsQueryWithState[*Executions](ctx, q.client, executionTable, combineToWhereStmt(query, queries.toQuery, eq), scan)
|
||||
}
|
||||
|
||||
func (q *Queries) GetExecutionByID(ctx context.Context, id string) (execution *Execution, err error) {
|
||||
eq := sq.Eq{
|
||||
ExecutionColumnID.identifier(): id,
|
||||
ExecutionColumnInstanceID.identifier(): authz.GetInstance(ctx).InstanceID(),
|
||||
}
|
||||
query, scan := prepareExecutionQuery(ctx, q.client)
|
||||
return genericRowQuery[*Execution](ctx, q.client, query.Where(eq), scan)
|
||||
}
|
||||
|
||||
func NewExecutionInIDsSearchQuery(values []string) (SearchQuery, error) {
|
||||
return NewInTextQuery(ExecutionColumnID, values)
|
||||
}
|
||||
|
||||
func NewExecutionTypeSearchQuery(t domain.ExecutionType) (SearchQuery, error) {
|
||||
return NewTextQuery(ExecutionColumnID, t.String(), TextStartsWith)
|
||||
}
|
||||
|
||||
func NewExecutionTargetSearchQuery(value string) (SearchQuery, error) {
|
||||
return NewTextQuery(ExecutionColumnTargets, value, TextListContains)
|
||||
}
|
||||
|
||||
func NewExecutionIncludeSearchQuery(value string) (SearchQuery, error) {
|
||||
return NewTextQuery(ExecutionColumnIncludes, value, TextListContains)
|
||||
}
|
||||
|
||||
func prepareExecutionsQuery(ctx context.Context, db prepareDatabase) (sq.SelectBuilder, func(rows *sql.Rows) (*Executions, error)) {
|
||||
return sq.Select(
|
||||
ExecutionColumnID.identifier(),
|
||||
ExecutionColumnChangeDate.identifier(),
|
||||
ExecutionColumnResourceOwner.identifier(),
|
||||
ExecutionColumnSequence.identifier(),
|
||||
ExecutionColumnTargets.identifier(),
|
||||
ExecutionColumnIncludes.identifier(),
|
||||
countColumn.identifier(),
|
||||
).From(executionTable.identifier()).
|
||||
PlaceholderFormat(sq.Dollar),
|
||||
func(rows *sql.Rows) (*Executions, error) {
|
||||
executions := make([]*Execution, 0)
|
||||
var count uint64
|
||||
for rows.Next() {
|
||||
execution := new(Execution)
|
||||
err := rows.Scan(
|
||||
&execution.ID,
|
||||
&execution.EventDate,
|
||||
&execution.ResourceOwner,
|
||||
&execution.Sequence,
|
||||
&execution.Targets,
|
||||
&execution.Includes,
|
||||
&count,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
executions = append(executions, execution)
|
||||
}
|
||||
|
||||
if err := rows.Close(); err != nil {
|
||||
return nil, zerrors.ThrowInternal(err, "QUERY-72xfx5jlj7", "Errors.Query.CloseRows")
|
||||
}
|
||||
|
||||
return &Executions{
|
||||
Executions: executions,
|
||||
SearchResponse: SearchResponse{
|
||||
Count: count,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
|
||||
func prepareExecutionQuery(ctx context.Context, db prepareDatabase) (sq.SelectBuilder, func(row *sql.Row) (*Execution, error)) {
|
||||
return sq.Select(
|
||||
ExecutionColumnID.identifier(),
|
||||
ExecutionColumnChangeDate.identifier(),
|
||||
ExecutionColumnResourceOwner.identifier(),
|
||||
ExecutionColumnSequence.identifier(),
|
||||
ExecutionColumnTargets.identifier(),
|
||||
ExecutionColumnIncludes.identifier(),
|
||||
).From(executionTable.identifier()).
|
||||
PlaceholderFormat(sq.Dollar),
|
||||
func(row *sql.Row) (*Execution, error) {
|
||||
execution := new(Execution)
|
||||
err := row.Scan(
|
||||
&execution.ID,
|
||||
&execution.EventDate,
|
||||
&execution.ResourceOwner,
|
||||
&execution.Sequence,
|
||||
&execution.Targets,
|
||||
&execution.Includes,
|
||||
)
|
||||
if err != nil {
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, zerrors.ThrowNotFound(err, "QUERY-qzn1xycesh", "Errors.Execution.NotFound")
|
||||
}
|
||||
return nil, zerrors.ThrowInternal(err, "QUERY-f8sjvm4tb8", "Errors.Internal")
|
||||
}
|
||||
return execution, nil
|
||||
}
|
||||
}
|
||||
253
internal/query/execution_test.go
Normal file
253
internal/query/execution_test.go
Normal file
@@ -0,0 +1,253 @@
|
||||
package query
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"database/sql/driver"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"testing"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/database"
|
||||
"github.com/zitadel/zitadel/internal/domain"
|
||||
"github.com/zitadel/zitadel/internal/zerrors"
|
||||
)
|
||||
|
||||
var (
|
||||
prepareExecutionsStmt = `SELECT projections.executions.id,` +
|
||||
` projections.executions.change_date,` +
|
||||
` projections.executions.resource_owner,` +
|
||||
` projections.executions.sequence,` +
|
||||
` projections.executions.targets,` +
|
||||
` projections.executions.includes,` +
|
||||
` COUNT(*) OVER ()` +
|
||||
` FROM projections.executions`
|
||||
prepareExecutionsCols = []string{
|
||||
"id",
|
||||
"change_date",
|
||||
"resource_owner",
|
||||
"sequence",
|
||||
"targets",
|
||||
"includes",
|
||||
"count",
|
||||
}
|
||||
|
||||
prepareExecutionStmt = `SELECT projections.executions.id,` +
|
||||
` projections.executions.change_date,` +
|
||||
` projections.executions.resource_owner,` +
|
||||
` projections.executions.sequence,` +
|
||||
` projections.executions.targets,` +
|
||||
` projections.executions.includes` +
|
||||
` FROM projections.executions`
|
||||
prepareExecutionCols = []string{
|
||||
"id",
|
||||
"change_date",
|
||||
"resource_owner",
|
||||
"sequence",
|
||||
"targets",
|
||||
"includes",
|
||||
}
|
||||
)
|
||||
|
||||
func Test_ExecutionPrepares(t *testing.T) {
|
||||
type want struct {
|
||||
sqlExpectations sqlExpectation
|
||||
err checkErr
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
prepare interface{}
|
||||
want want
|
||||
object interface{}
|
||||
}{
|
||||
{
|
||||
name: "prepareExecutionsQuery no result",
|
||||
prepare: prepareExecutionsQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueries(
|
||||
regexp.QuoteMeta(prepareExecutionsStmt),
|
||||
nil,
|
||||
nil,
|
||||
),
|
||||
},
|
||||
object: &Executions{Executions: []*Execution{}},
|
||||
},
|
||||
{
|
||||
name: "prepareExecutionsQuery one result",
|
||||
prepare: prepareExecutionsQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueries(
|
||||
regexp.QuoteMeta(prepareExecutionsStmt),
|
||||
prepareExecutionsCols,
|
||||
[][]driver.Value{
|
||||
{
|
||||
"id",
|
||||
testNow,
|
||||
"ro",
|
||||
uint64(20211109),
|
||||
database.TextArray[string]{"target"},
|
||||
database.TextArray[string]{"include"},
|
||||
},
|
||||
},
|
||||
),
|
||||
},
|
||||
object: &Executions{
|
||||
SearchResponse: SearchResponse{
|
||||
Count: 1,
|
||||
},
|
||||
Executions: []*Execution{
|
||||
{
|
||||
ID: "id",
|
||||
ObjectDetails: domain.ObjectDetails{
|
||||
EventDate: testNow,
|
||||
ResourceOwner: "ro",
|
||||
Sequence: 20211109,
|
||||
},
|
||||
Targets: database.TextArray[string]{"target"},
|
||||
Includes: database.TextArray[string]{"include"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "prepareExecutionsQuery multiple result",
|
||||
prepare: prepareExecutionsQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueries(
|
||||
regexp.QuoteMeta(prepareExecutionsStmt),
|
||||
prepareExecutionsCols,
|
||||
[][]driver.Value{
|
||||
{
|
||||
"id-1",
|
||||
testNow,
|
||||
"ro",
|
||||
uint64(20211109),
|
||||
database.TextArray[string]{"target1"},
|
||||
database.TextArray[string]{"include1"},
|
||||
},
|
||||
{
|
||||
"id-2",
|
||||
testNow,
|
||||
"ro",
|
||||
uint64(20211110),
|
||||
database.TextArray[string]{"target2"},
|
||||
database.TextArray[string]{"include2"},
|
||||
},
|
||||
},
|
||||
),
|
||||
},
|
||||
object: &Executions{
|
||||
SearchResponse: SearchResponse{
|
||||
Count: 2,
|
||||
},
|
||||
Executions: []*Execution{
|
||||
{
|
||||
ID: "id-1",
|
||||
ObjectDetails: domain.ObjectDetails{
|
||||
EventDate: testNow,
|
||||
ResourceOwner: "ro",
|
||||
Sequence: 20211109,
|
||||
},
|
||||
Targets: database.TextArray[string]{"target1"},
|
||||
Includes: database.TextArray[string]{"include1"},
|
||||
},
|
||||
{
|
||||
ID: "id-2",
|
||||
ObjectDetails: domain.ObjectDetails{
|
||||
EventDate: testNow,
|
||||
ResourceOwner: "ro",
|
||||
Sequence: 20211110,
|
||||
},
|
||||
Targets: database.TextArray[string]{"target2"},
|
||||
Includes: database.TextArray[string]{"include2"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "prepareExecutionsQuery sql err",
|
||||
prepare: prepareExecutionsQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueryErr(
|
||||
regexp.QuoteMeta(prepareExecutionsStmt),
|
||||
sql.ErrConnDone,
|
||||
),
|
||||
err: func(err error) (error, bool) {
|
||||
if !errors.Is(err, sql.ErrConnDone) {
|
||||
return fmt.Errorf("err should be sql.ErrConnDone got: %w", err), false
|
||||
}
|
||||
return nil, true
|
||||
},
|
||||
},
|
||||
object: (*Execution)(nil),
|
||||
},
|
||||
{
|
||||
name: "prepareExecutionQuery no result",
|
||||
prepare: prepareExecutionQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueriesScanErr(
|
||||
regexp.QuoteMeta(prepareExecutionStmt),
|
||||
nil,
|
||||
nil,
|
||||
),
|
||||
err: func(err error) (error, bool) {
|
||||
if !zerrors.IsNotFound(err) {
|
||||
return fmt.Errorf("err should be zitadel.NotFoundError got: %w", err), false
|
||||
}
|
||||
return nil, true
|
||||
},
|
||||
},
|
||||
object: (*Execution)(nil),
|
||||
},
|
||||
{
|
||||
name: "prepareExecutionQuery found",
|
||||
prepare: prepareExecutionQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQuery(
|
||||
regexp.QuoteMeta(prepareExecutionStmt),
|
||||
prepareExecutionCols,
|
||||
[]driver.Value{
|
||||
"id",
|
||||
testNow,
|
||||
"ro",
|
||||
uint64(20211109),
|
||||
database.TextArray[string]{"target"},
|
||||
database.TextArray[string]{"include"},
|
||||
},
|
||||
),
|
||||
},
|
||||
object: &Execution{
|
||||
ID: "id",
|
||||
ObjectDetails: domain.ObjectDetails{
|
||||
EventDate: testNow,
|
||||
ResourceOwner: "ro",
|
||||
Sequence: 20211109,
|
||||
},
|
||||
Targets: database.TextArray[string]{"target"},
|
||||
Includes: database.TextArray[string]{"include"},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "prepareExecutionQuery sql err",
|
||||
prepare: prepareExecutionQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueryErr(
|
||||
regexp.QuoteMeta(prepareExecutionStmt),
|
||||
sql.ErrConnDone,
|
||||
),
|
||||
err: func(err error) (error, bool) {
|
||||
if !errors.Is(err, sql.ErrConnDone) {
|
||||
return fmt.Errorf("err should be sql.ErrConnDone got: %w", err), false
|
||||
}
|
||||
return nil, true
|
||||
},
|
||||
},
|
||||
object: (*Execution)(nil),
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
assertPrepare(t, tt.prepare, tt.object, tt.want.sqlExpectations, tt.want.err, defaultPrepareArgs...)
|
||||
})
|
||||
}
|
||||
}
|
||||
109
internal/query/generic.go
Normal file
109
internal/query/generic.go
Normal file
@@ -0,0 +1,109 @@
|
||||
package query
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
|
||||
sq "github.com/Masterminds/squirrel"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/api/authz"
|
||||
"github.com/zitadel/zitadel/internal/database"
|
||||
"github.com/zitadel/zitadel/internal/telemetry/tracing"
|
||||
"github.com/zitadel/zitadel/internal/zerrors"
|
||||
)
|
||||
|
||||
func genericRowsQuery[R any](
|
||||
ctx context.Context,
|
||||
client *database.DB,
|
||||
query sq.SelectBuilder,
|
||||
scan func(rows *sql.Rows) (R, error),
|
||||
) (resp R, err error) {
|
||||
var rnil R
|
||||
ctx, span := tracing.NewSpan(ctx)
|
||||
defer func() { span.EndWithError(err) }()
|
||||
|
||||
stmt, args, err := query.ToSql()
|
||||
if err != nil {
|
||||
return rnil, zerrors.ThrowInvalidArgument(err, "QUERY-05wf2q36ji", "Errors.Query.InvalidRequest")
|
||||
}
|
||||
err = client.QueryContext(ctx, func(rows *sql.Rows) error {
|
||||
resp, err = scan(rows)
|
||||
return err
|
||||
}, stmt, args...)
|
||||
if err != nil {
|
||||
return rnil, zerrors.ThrowInternal(err, "QUERY-y2u7vctrha", "Errors.Internal")
|
||||
}
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func genericRowsQueryWithState[R Stateful](
|
||||
ctx context.Context,
|
||||
client *database.DB,
|
||||
projection table,
|
||||
query sq.SelectBuilder,
|
||||
scan func(rows *sql.Rows) (R, error),
|
||||
) (resp R, err error) {
|
||||
var rnil R
|
||||
resp, err = genericRowsQuery[R](ctx, client, query, scan)
|
||||
if err != nil {
|
||||
return rnil, err
|
||||
}
|
||||
state, err := latestState(ctx, client, projection)
|
||||
if err != nil {
|
||||
return rnil, err
|
||||
}
|
||||
resp.SetState(state)
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func latestState(ctx context.Context, client *database.DB, projections ...table) (state *State, err error) {
|
||||
ctx, span := tracing.NewSpan(ctx)
|
||||
defer func() { span.EndWithError(err) }()
|
||||
|
||||
query, scan := prepareLatestState(ctx, client)
|
||||
or := make(sq.Or, len(projections))
|
||||
for i, projection := range projections {
|
||||
or[i] = sq.Eq{CurrentStateColProjectionName.identifier(): projection.name}
|
||||
}
|
||||
stmt, args, err := query.
|
||||
Where(or).
|
||||
Where(sq.Eq{CurrentStateColInstanceID.identifier(): authz.GetInstance(ctx).InstanceID()}).
|
||||
OrderBy(CurrentStateColEventDate.identifier() + " DESC").
|
||||
ToSql()
|
||||
if err != nil {
|
||||
return nil, zerrors.ThrowInternal(err, "QUERY-5CfX9", "Errors.Query.SQLStatement")
|
||||
}
|
||||
|
||||
err = client.QueryRowContext(ctx, func(row *sql.Row) error {
|
||||
state, err = scan(row)
|
||||
return err
|
||||
}, stmt, args...)
|
||||
|
||||
return state, err
|
||||
}
|
||||
|
||||
func genericRowQuery[R any](
|
||||
ctx context.Context,
|
||||
client *database.DB,
|
||||
query sq.SelectBuilder,
|
||||
scan func(row *sql.Row) (R, error),
|
||||
) (resp R, err error) {
|
||||
var rnil R
|
||||
ctx, span := tracing.NewSpan(ctx)
|
||||
defer func() { span.EndWithError(err) }()
|
||||
|
||||
stmt, args, err := query.ToSql()
|
||||
if err != nil {
|
||||
return rnil, zerrors.ThrowInternal(err, "QUERY-s969t763z4", "Errors.Query.SQLStatement")
|
||||
}
|
||||
|
||||
err = client.QueryRowContext(ctx, func(row *sql.Row) error {
|
||||
resp, err = scan(row)
|
||||
return err
|
||||
}, stmt, args...)
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func combineToWhereStmt(query sq.SelectBuilder, toQuery func(query sq.SelectBuilder) sq.SelectBuilder, eq interface{}) sq.SelectBuilder {
|
||||
return toQuery(query).Where(eq)
|
||||
}
|
||||
@@ -99,7 +99,7 @@ func TestActionProjection_reduces(t *testing.T) {
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
action.ChangedEventType,
|
||||
action.DeactivatedEventType,
|
||||
action.AggregateType,
|
||||
[]byte(`{}`),
|
||||
),
|
||||
@@ -131,7 +131,7 @@ func TestActionProjection_reduces(t *testing.T) {
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
action.ChangedEventType,
|
||||
action.ReactivatedEventType,
|
||||
action.AggregateType,
|
||||
[]byte(`{}`),
|
||||
),
|
||||
@@ -163,7 +163,7 @@ func TestActionProjection_reduces(t *testing.T) {
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
action.ChangedEventType,
|
||||
action.RemovedEventType,
|
||||
action.AggregateType,
|
||||
[]byte(`{}`),
|
||||
),
|
||||
|
||||
109
internal/query/projection/execution.go
Normal file
109
internal/query/projection/execution.go
Normal file
@@ -0,0 +1,109 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/eventstore"
|
||||
old_handler "github.com/zitadel/zitadel/internal/eventstore/handler"
|
||||
"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
|
||||
exec "github.com/zitadel/zitadel/internal/repository/execution"
|
||||
"github.com/zitadel/zitadel/internal/repository/instance"
|
||||
)
|
||||
|
||||
const (
|
||||
ExecutionTable = "projections.executions"
|
||||
ExecutionIDCol = "id"
|
||||
ExecutionCreationDateCol = "creation_date"
|
||||
ExecutionChangeDateCol = "change_date"
|
||||
ExecutionResourceOwnerCol = "resource_owner"
|
||||
ExecutionInstanceIDCol = "instance_id"
|
||||
ExecutionSequenceCol = "sequence"
|
||||
ExecutionTargetsCol = "targets"
|
||||
ExecutionIncludesCol = "includes"
|
||||
)
|
||||
|
||||
type executionProjection struct{}
|
||||
|
||||
func newExecutionProjection(ctx context.Context, config handler.Config) *handler.Handler {
|
||||
return handler.NewHandler(ctx, &config, new(executionProjection))
|
||||
}
|
||||
|
||||
func (*executionProjection) Name() string {
|
||||
return ExecutionTable
|
||||
}
|
||||
|
||||
func (*executionProjection) Init() *old_handler.Check {
|
||||
return handler.NewTableCheck(
|
||||
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.NewPrimaryKey(ExecutionInstanceIDCol, ExecutionIDCol),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
func (p *executionProjection) Reducers() []handler.AggregateReducer {
|
||||
return []handler.AggregateReducer{
|
||||
{
|
||||
Aggregate: exec.AggregateType,
|
||||
EventReducers: []handler.EventReducer{
|
||||
{
|
||||
Event: exec.SetEventType,
|
||||
Reduce: p.reduceExecutionSet,
|
||||
},
|
||||
{
|
||||
Event: exec.RemovedEventType,
|
||||
Reduce: p.reduceExecutionRemoved,
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Aggregate: instance.AggregateType,
|
||||
EventReducers: []handler.EventReducer{
|
||||
{
|
||||
Event: instance.InstanceRemovedEventType,
|
||||
Reduce: reduceInstanceRemovedHelper(ExecutionInstanceIDCol),
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (p *executionProjection) reduceExecutionSet(event eventstore.Event) (*handler.Statement, error) {
|
||||
e, err := assertEvent[*exec.SetEvent](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),
|
||||
}
|
||||
return handler.NewUpsertStatement(e, columns[0:2], columns), nil
|
||||
}
|
||||
|
||||
func (p *executionProjection) reduceExecutionRemoved(event eventstore.Event) (*handler.Statement, error) {
|
||||
e, err := assertEvent[*exec.RemovedEvent](event)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return handler.NewDeleteStatement(
|
||||
e,
|
||||
[]handler.Condition{
|
||||
handler.NewCond(ExecutionInstanceIDCol, e.Aggregate().InstanceID),
|
||||
handler.NewCond(ExecutionIDCol, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
129
internal/query/projection/execution_test.go
Normal file
129
internal/query/projection/execution_test.go
Normal file
@@ -0,0 +1,129 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/eventstore"
|
||||
"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
|
||||
exec "github.com/zitadel/zitadel/internal/repository/execution"
|
||||
"github.com/zitadel/zitadel/internal/repository/instance"
|
||||
"github.com/zitadel/zitadel/internal/zerrors"
|
||||
)
|
||||
|
||||
func TestExecutionProjection_reduces(t *testing.T) {
|
||||
type args struct {
|
||||
event func(t *testing.T) eventstore.Event
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
reduce func(event eventstore.Event) (*handler.Statement, error)
|
||||
want wantReduce
|
||||
}{
|
||||
{
|
||||
name: "reduceExecutionSet",
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
exec.SetEventType,
|
||||
exec.AggregateType,
|
||||
[]byte(`{"targets": ["target"], "includes": ["include"]}`),
|
||||
),
|
||||
eventstore.GenericEventMapper[exec.SetEvent],
|
||||
),
|
||||
},
|
||||
reduce: (&executionProjection{}).reduceExecutionSet,
|
||||
want: wantReduce{
|
||||
aggregateType: eventstore.AggregateType("execution"),
|
||||
sequence: 15,
|
||||
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)",
|
||||
expectedArgs: []interface{}{
|
||||
"instance-id",
|
||||
"agg-id",
|
||||
"ro-id",
|
||||
anyArg{},
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
[]string{"target"},
|
||||
[]string{"include"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceExecutionRemoved",
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
exec.RemovedEventType,
|
||||
exec.AggregateType,
|
||||
[]byte(`{}`),
|
||||
),
|
||||
eventstore.GenericEventMapper[exec.RemovedEvent],
|
||||
),
|
||||
},
|
||||
reduce: (&executionProjection{}).reduceExecutionRemoved,
|
||||
want: wantReduce{
|
||||
aggregateType: eventstore.AggregateType("execution"),
|
||||
sequence: 15,
|
||||
executer: &testExecuter{
|
||||
executions: []execution{
|
||||
{
|
||||
expectedStmt: "DELETE FROM projections.executions WHERE (instance_id = $1) AND (id = $2)",
|
||||
expectedArgs: []interface{}{
|
||||
"instance-id",
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceInstanceRemoved",
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
instance.InstanceRemovedEventType,
|
||||
instance.AggregateType,
|
||||
nil,
|
||||
),
|
||||
instance.InstanceRemovedEventMapper,
|
||||
),
|
||||
},
|
||||
reduce: reduceInstanceRemovedHelper(ExecutionInstanceIDCol),
|
||||
want: wantReduce{
|
||||
aggregateType: eventstore.AggregateType("instance"),
|
||||
sequence: 15,
|
||||
executer: &testExecuter{
|
||||
executions: []execution{
|
||||
{
|
||||
expectedStmt: "DELETE FROM projections.executions WHERE (instance_id = $1)",
|
||||
expectedArgs: []interface{}{
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
event := baseEvent(t)
|
||||
got, err := tt.reduce(event)
|
||||
if ok := zerrors.IsErrorInvalidArgument(err); !ok {
|
||||
t.Errorf("no wrong event mapping: %v, got: %v", err, got)
|
||||
}
|
||||
|
||||
event = tt.args.event(t)
|
||||
got, err = tt.reduce(event)
|
||||
assertReduce(t, got, err, ExecutionTable, tt.want)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -74,6 +74,8 @@ var (
|
||||
RestrictionsProjection *handler.Handler
|
||||
SystemFeatureProjection *handler.Handler
|
||||
InstanceFeatureProjection *handler.Handler
|
||||
TargetProjection *handler.Handler
|
||||
ExecutionProjection *handler.Handler
|
||||
)
|
||||
|
||||
type projection interface {
|
||||
@@ -152,6 +154,8 @@ func Create(ctx context.Context, sqlClient *database.DB, es handler.EventStore,
|
||||
RestrictionsProjection = newRestrictionsProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["restrictions"]))
|
||||
SystemFeatureProjection = newSystemFeatureProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["system_features"]))
|
||||
InstanceFeatureProjection = newInstanceFeatureProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["instance_features"]))
|
||||
TargetProjection = newTargetProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["targets"]))
|
||||
ExecutionProjection = newExecutionProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["executions"]))
|
||||
newProjectionsList()
|
||||
return nil
|
||||
}
|
||||
@@ -263,5 +267,7 @@ func newProjectionsList() {
|
||||
RestrictionsProjection,
|
||||
SystemFeatureProjection,
|
||||
InstanceFeatureProjection,
|
||||
ExecutionProjection,
|
||||
TargetProjection,
|
||||
}
|
||||
}
|
||||
|
||||
165
internal/query/projection/target.go
Normal file
165
internal/query/projection/target.go
Normal file
@@ -0,0 +1,165 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/eventstore"
|
||||
old_handler "github.com/zitadel/zitadel/internal/eventstore/handler"
|
||||
"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
|
||||
"github.com/zitadel/zitadel/internal/repository/instance"
|
||||
"github.com/zitadel/zitadel/internal/repository/target"
|
||||
)
|
||||
|
||||
const (
|
||||
TargetTable = "projections.targets"
|
||||
TargetIDCol = "id"
|
||||
TargetCreationDateCol = "creation_date"
|
||||
TargetChangeDateCol = "change_date"
|
||||
TargetResourceOwnerCol = "resource_owner"
|
||||
TargetInstanceIDCol = "instance_id"
|
||||
TargetSequenceCol = "sequence"
|
||||
TargetNameCol = "name"
|
||||
TargetTargetType = "target_type"
|
||||
TargetURLCol = "url"
|
||||
TargetTimeoutCol = "timeout"
|
||||
TargetAsyncCol = "async"
|
||||
TargetInterruptOnErrorCol = "interrupt_on_error"
|
||||
)
|
||||
|
||||
type targetProjection struct{}
|
||||
|
||||
func newTargetProjection(ctx context.Context, config handler.Config) *handler.Handler {
|
||||
return handler.NewHandler(ctx, &config, new(targetProjection))
|
||||
}
|
||||
|
||||
func (*targetProjection) Name() string {
|
||||
return TargetTable
|
||||
}
|
||||
|
||||
func (*targetProjection) Init() *old_handler.Check {
|
||||
return handler.NewTableCheck(
|
||||
handler.NewTable([]*handler.InitColumn{
|
||||
handler.NewColumn(TargetIDCol, handler.ColumnTypeText),
|
||||
handler.NewColumn(TargetCreationDateCol, handler.ColumnTypeTimestamp),
|
||||
handler.NewColumn(TargetChangeDateCol, handler.ColumnTypeTimestamp),
|
||||
handler.NewColumn(TargetResourceOwnerCol, handler.ColumnTypeText),
|
||||
handler.NewColumn(TargetInstanceIDCol, handler.ColumnTypeText),
|
||||
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.NewPrimaryKey(TargetInstanceIDCol, TargetIDCol),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
func (p *targetProjection) Reducers() []handler.AggregateReducer {
|
||||
return []handler.AggregateReducer{
|
||||
{
|
||||
Aggregate: target.AggregateType,
|
||||
EventReducers: []handler.EventReducer{
|
||||
{
|
||||
Event: target.AddedEventType,
|
||||
Reduce: p.reduceTargetAdded,
|
||||
},
|
||||
{
|
||||
Event: target.ChangedEventType,
|
||||
Reduce: p.reduceTargetChanged,
|
||||
},
|
||||
{
|
||||
Event: target.RemovedEventType,
|
||||
Reduce: p.reduceTargetRemoved,
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Aggregate: instance.AggregateType,
|
||||
EventReducers: []handler.EventReducer{
|
||||
{
|
||||
Event: instance.InstanceRemovedEventType,
|
||||
Reduce: reduceInstanceRemovedHelper(TargetInstanceIDCol),
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (p *targetProjection) reduceTargetAdded(event eventstore.Event) (*handler.Statement, error) {
|
||||
e, err := assertEvent[*target.AddedEvent](event)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return handler.NewCreateStatement(
|
||||
e,
|
||||
[]handler.Column{
|
||||
handler.NewCol(TargetInstanceIDCol, e.Aggregate().InstanceID),
|
||||
handler.NewCol(TargetResourceOwnerCol, e.Aggregate().ResourceOwner),
|
||||
handler.NewCol(TargetIDCol, e.Aggregate().ID),
|
||||
handler.NewCol(TargetCreationDateCol, e.CreationDate()),
|
||||
handler.NewCol(TargetChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(TargetSequenceCol, e.Sequence()),
|
||||
handler.NewCol(TargetNameCol, e.Name),
|
||||
handler.NewCol(TargetURLCol, e.URL),
|
||||
handler.NewCol(TargetTargetType, e.TargetType),
|
||||
handler.NewCol(TargetTimeoutCol, e.Timeout),
|
||||
handler.NewCol(TargetAsyncCol, e.Async),
|
||||
handler.NewCol(TargetInterruptOnErrorCol, e.InterruptOnError),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
|
||||
func (p *targetProjection) reduceTargetChanged(event eventstore.Event) (*handler.Statement, error) {
|
||||
e, err := assertEvent[*target.ChangedEvent](event)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
values := []handler.Column{
|
||||
handler.NewCol(TargetChangeDateCol, e.CreationDate()),
|
||||
handler.NewCol(TargetSequenceCol, e.Sequence()),
|
||||
handler.NewCol(TargetResourceOwnerCol, e.Aggregate().ResourceOwner),
|
||||
}
|
||||
if e.Name != nil {
|
||||
values = append(values, handler.NewCol(TargetNameCol, *e.Name))
|
||||
}
|
||||
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.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))
|
||||
}
|
||||
return handler.NewUpdateStatement(
|
||||
e,
|
||||
values,
|
||||
[]handler.Condition{
|
||||
handler.NewCond(TargetInstanceIDCol, e.Aggregate().InstanceID),
|
||||
handler.NewCond(TargetIDCol, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
|
||||
func (p *targetProjection) reduceTargetRemoved(event eventstore.Event) (*handler.Statement, error) {
|
||||
e, err := assertEvent[*target.RemovedEvent](event)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return handler.NewDeleteStatement(
|
||||
e,
|
||||
[]handler.Condition{
|
||||
handler.NewCond(TargetInstanceIDCol, e.Aggregate().InstanceID),
|
||||
handler.NewCond(TargetIDCol, e.Aggregate().ID),
|
||||
},
|
||||
), nil
|
||||
}
|
||||
173
internal/query/projection/target_test.go
Normal file
173
internal/query/projection/target_test.go
Normal file
@@ -0,0 +1,173 @@
|
||||
package projection
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/domain"
|
||||
"github.com/zitadel/zitadel/internal/eventstore"
|
||||
"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
|
||||
"github.com/zitadel/zitadel/internal/repository/instance"
|
||||
"github.com/zitadel/zitadel/internal/repository/target"
|
||||
"github.com/zitadel/zitadel/internal/zerrors"
|
||||
)
|
||||
|
||||
func TestTargetProjection_reduces(t *testing.T) {
|
||||
type args struct {
|
||||
event func(t *testing.T) eventstore.Event
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
reduce func(event eventstore.Event) (*handler.Statement, error)
|
||||
want wantReduce
|
||||
}{
|
||||
{
|
||||
name: "reduceTargetAdded",
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
target.AddedEventType,
|
||||
target.AggregateType,
|
||||
[]byte(`{"name": "name", "targetType":0, "url":"https://example.com", "timeout": 3000000000, "async": true, "interruptOnError": true}`),
|
||||
),
|
||||
eventstore.GenericEventMapper[target.AddedEvent],
|
||||
),
|
||||
},
|
||||
reduce: (&targetProjection{}).reduceTargetAdded,
|
||||
want: wantReduce{
|
||||
aggregateType: eventstore.AggregateType("target"),
|
||||
sequence: 15,
|
||||
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)",
|
||||
expectedArgs: []interface{}{
|
||||
"instance-id",
|
||||
"ro-id",
|
||||
"agg-id",
|
||||
anyArg{},
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
"name",
|
||||
"https://example.com",
|
||||
domain.TargetTypeWebhook,
|
||||
3 * time.Second,
|
||||
true,
|
||||
true,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceTargetChanged",
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
target.ChangedEventType,
|
||||
target.AggregateType,
|
||||
[]byte(`{"name": "name2", "targetType":0, "url":"https://example.com", "timeout": 3000000000, "async": true, "interruptOnError": true}`),
|
||||
),
|
||||
eventstore.GenericEventMapper[target.ChangedEvent],
|
||||
),
|
||||
},
|
||||
reduce: (&targetProjection{}).reduceTargetChanged,
|
||||
want: wantReduce{
|
||||
aggregateType: eventstore.AggregateType("target"),
|
||||
sequence: 15,
|
||||
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)",
|
||||
expectedArgs: []interface{}{
|
||||
anyArg{},
|
||||
uint64(15),
|
||||
"ro-id",
|
||||
"name2",
|
||||
domain.TargetTypeWebhook,
|
||||
"https://example.com",
|
||||
3 * time.Second,
|
||||
true,
|
||||
true,
|
||||
"instance-id",
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceTargetRemoved",
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
target.RemovedEventType,
|
||||
target.AggregateType,
|
||||
[]byte(`{}`),
|
||||
),
|
||||
eventstore.GenericEventMapper[target.RemovedEvent],
|
||||
),
|
||||
},
|
||||
reduce: (&targetProjection{}).reduceTargetRemoved,
|
||||
want: wantReduce{
|
||||
aggregateType: eventstore.AggregateType("target"),
|
||||
sequence: 15,
|
||||
executer: &testExecuter{
|
||||
executions: []execution{
|
||||
{
|
||||
expectedStmt: "DELETE FROM projections.targets WHERE (instance_id = $1) AND (id = $2)",
|
||||
expectedArgs: []interface{}{
|
||||
"instance-id",
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "reduceInstanceRemoved",
|
||||
args: args{
|
||||
event: getEvent(
|
||||
testEvent(
|
||||
instance.InstanceRemovedEventType,
|
||||
instance.AggregateType,
|
||||
nil,
|
||||
),
|
||||
instance.InstanceRemovedEventMapper,
|
||||
),
|
||||
},
|
||||
reduce: reduceInstanceRemovedHelper(TargetInstanceIDCol),
|
||||
want: wantReduce{
|
||||
aggregateType: eventstore.AggregateType("instance"),
|
||||
sequence: 15,
|
||||
executer: &testExecuter{
|
||||
executions: []execution{
|
||||
{
|
||||
expectedStmt: "DELETE FROM projections.targets WHERE (instance_id = $1)",
|
||||
expectedArgs: []interface{}{
|
||||
"agg-id",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
event := baseEvent(t)
|
||||
got, err := tt.reduce(event)
|
||||
if ok := zerrors.IsErrorInvalidArgument(err); !ok {
|
||||
t.Errorf("no wrong event mapping: %v, got: %v", err, got)
|
||||
}
|
||||
|
||||
event = tt.args.event(t)
|
||||
got, err = tt.reduce(event)
|
||||
assertReduce(t, got, err, TargetTable, tt.want)
|
||||
})
|
||||
}
|
||||
}
|
||||
219
internal/query/target.go
Normal file
219
internal/query/target.go
Normal file
@@ -0,0 +1,219 @@
|
||||
package query
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
sq "github.com/Masterminds/squirrel"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/api/authz"
|
||||
"github.com/zitadel/zitadel/internal/domain"
|
||||
"github.com/zitadel/zitadel/internal/query/projection"
|
||||
"github.com/zitadel/zitadel/internal/zerrors"
|
||||
)
|
||||
|
||||
var (
|
||||
targetTable = table{
|
||||
name: projection.TargetTable,
|
||||
instanceIDCol: projection.TargetInstanceIDCol,
|
||||
}
|
||||
TargetColumnID = Column{
|
||||
name: projection.TargetIDCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnCreationDate = Column{
|
||||
name: projection.TargetCreationDateCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnChangeDate = Column{
|
||||
name: projection.TargetChangeDateCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnResourceOwner = Column{
|
||||
name: projection.TargetResourceOwnerCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnInstanceID = Column{
|
||||
name: projection.TargetInstanceIDCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnSequence = Column{
|
||||
name: projection.TargetSequenceCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnName = Column{
|
||||
name: projection.TargetNameCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnTargetType = Column{
|
||||
name: projection.TargetTargetType,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnURL = Column{
|
||||
name: projection.TargetURLCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnTimeout = Column{
|
||||
name: projection.TargetTimeoutCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnAsync = Column{
|
||||
name: projection.TargetAsyncCol,
|
||||
table: targetTable,
|
||||
}
|
||||
TargetColumnInterruptOnError = Column{
|
||||
name: projection.TargetInterruptOnErrorCol,
|
||||
table: targetTable,
|
||||
}
|
||||
)
|
||||
|
||||
type Targets struct {
|
||||
SearchResponse
|
||||
Targets []*Target
|
||||
}
|
||||
|
||||
func (t *Targets) SetState(s *State) {
|
||||
t.State = s
|
||||
}
|
||||
|
||||
type Target struct {
|
||||
ID string
|
||||
domain.ObjectDetails
|
||||
|
||||
Name string
|
||||
TargetType domain.TargetType
|
||||
URL string
|
||||
Timeout time.Duration
|
||||
Async bool
|
||||
InterruptOnError bool
|
||||
}
|
||||
|
||||
type TargetSearchQueries struct {
|
||||
SearchRequest
|
||||
Queries []SearchQuery
|
||||
}
|
||||
|
||||
func (q *TargetSearchQueries) toQuery(query sq.SelectBuilder) sq.SelectBuilder {
|
||||
query = q.SearchRequest.toQuery(query)
|
||||
for _, q := range q.Queries {
|
||||
query = q.toQuery(query)
|
||||
}
|
||||
return query
|
||||
}
|
||||
|
||||
func (q *Queries) SearchTargets(ctx context.Context, queries *TargetSearchQueries) (targets *Targets, err error) {
|
||||
eq := sq.Eq{
|
||||
TargetColumnInstanceID.identifier(): authz.GetInstance(ctx).InstanceID(),
|
||||
}
|
||||
query, scan := prepareTargetsQuery(ctx, q.client)
|
||||
return genericRowsQueryWithState[*Targets](ctx, q.client, targetTable, combineToWhereStmt(query, queries.toQuery, eq), scan)
|
||||
}
|
||||
|
||||
func (q *Queries) GetTargetByID(ctx context.Context, id string) (target *Target, err error) {
|
||||
eq := sq.Eq{
|
||||
TargetColumnID.identifier(): id,
|
||||
TargetColumnInstanceID.identifier(): authz.GetInstance(ctx).InstanceID(),
|
||||
}
|
||||
query, scan := prepareTargetQuery(ctx, q.client)
|
||||
return genericRowQuery[*Target](ctx, q.client, query.Where(eq), scan)
|
||||
}
|
||||
|
||||
func NewTargetNameSearchQuery(method TextComparison, value string) (SearchQuery, error) {
|
||||
return NewTextQuery(TargetColumnName, value, method)
|
||||
}
|
||||
|
||||
func NewTargetInIDsSearchQuery(values []string) (SearchQuery, error) {
|
||||
return NewInTextQuery(TargetColumnID, values)
|
||||
}
|
||||
|
||||
func prepareTargetsQuery(ctx context.Context, db prepareDatabase) (sq.SelectBuilder, func(rows *sql.Rows) (*Targets, error)) {
|
||||
return sq.Select(
|
||||
TargetColumnID.identifier(),
|
||||
TargetColumnChangeDate.identifier(),
|
||||
TargetColumnResourceOwner.identifier(),
|
||||
TargetColumnSequence.identifier(),
|
||||
TargetColumnName.identifier(),
|
||||
TargetColumnTargetType.identifier(),
|
||||
TargetColumnTimeout.identifier(),
|
||||
TargetColumnURL.identifier(),
|
||||
TargetColumnAsync.identifier(),
|
||||
TargetColumnInterruptOnError.identifier(),
|
||||
countColumn.identifier(),
|
||||
).From(targetTable.identifier()).
|
||||
PlaceholderFormat(sq.Dollar),
|
||||
func(rows *sql.Rows) (*Targets, error) {
|
||||
targets := make([]*Target, 0)
|
||||
var count uint64
|
||||
for rows.Next() {
|
||||
target := new(Target)
|
||||
err := rows.Scan(
|
||||
&target.ID,
|
||||
&target.EventDate,
|
||||
&target.ResourceOwner,
|
||||
&target.Sequence,
|
||||
&target.Name,
|
||||
&target.TargetType,
|
||||
&target.Timeout,
|
||||
&target.URL,
|
||||
&target.Async,
|
||||
&target.InterruptOnError,
|
||||
&count,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
targets = append(targets, target)
|
||||
}
|
||||
|
||||
if err := rows.Close(); err != nil {
|
||||
return nil, zerrors.ThrowInternal(err, "QUERY-fzwi6cgxos", "Errors.Query.CloseRows")
|
||||
}
|
||||
|
||||
return &Targets{
|
||||
Targets: targets,
|
||||
SearchResponse: SearchResponse{
|
||||
Count: count,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
|
||||
func prepareTargetQuery(ctx context.Context, db prepareDatabase) (sq.SelectBuilder, func(row *sql.Row) (*Target, error)) {
|
||||
return sq.Select(
|
||||
TargetColumnID.identifier(),
|
||||
TargetColumnChangeDate.identifier(),
|
||||
TargetColumnResourceOwner.identifier(),
|
||||
TargetColumnSequence.identifier(),
|
||||
TargetColumnName.identifier(),
|
||||
TargetColumnTargetType.identifier(),
|
||||
TargetColumnTimeout.identifier(),
|
||||
TargetColumnURL.identifier(),
|
||||
TargetColumnAsync.identifier(),
|
||||
TargetColumnInterruptOnError.identifier(),
|
||||
).From(targetTable.identifier()).
|
||||
PlaceholderFormat(sq.Dollar),
|
||||
func(row *sql.Row) (*Target, error) {
|
||||
target := new(Target)
|
||||
err := row.Scan(
|
||||
&target.ID,
|
||||
&target.EventDate,
|
||||
&target.ResourceOwner,
|
||||
&target.Sequence,
|
||||
&target.Name,
|
||||
&target.TargetType,
|
||||
&target.Timeout,
|
||||
&target.URL,
|
||||
&target.Async,
|
||||
&target.InterruptOnError,
|
||||
)
|
||||
if err != nil {
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, zerrors.ThrowNotFound(err, "QUERY-hj5oaniyrz", "Errors.Target.NotFound")
|
||||
}
|
||||
return nil, zerrors.ThrowInternal(err, "QUERY-5qhc19sc49", "Errors.Internal")
|
||||
}
|
||||
return target, nil
|
||||
}
|
||||
}
|
||||
301
internal/query/target_test.go
Normal file
301
internal/query/target_test.go
Normal file
@@ -0,0 +1,301 @@
|
||||
package query
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"database/sql/driver"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/domain"
|
||||
"github.com/zitadel/zitadel/internal/zerrors"
|
||||
)
|
||||
|
||||
var (
|
||||
prepareTargetsStmt = `SELECT projections.targets.id,` +
|
||||
` projections.targets.change_date,` +
|
||||
` projections.targets.resource_owner,` +
|
||||
` projections.targets.sequence,` +
|
||||
` projections.targets.name,` +
|
||||
` projections.targets.target_type,` +
|
||||
` projections.targets.timeout,` +
|
||||
` projections.targets.url,` +
|
||||
` projections.targets.async,` +
|
||||
` projections.targets.interrupt_on_error,` +
|
||||
` COUNT(*) OVER ()` +
|
||||
` FROM projections.targets`
|
||||
prepareTargetsCols = []string{
|
||||
"id",
|
||||
"change_date",
|
||||
"resource_owner",
|
||||
"sequence",
|
||||
"name",
|
||||
"target_type",
|
||||
"timeout",
|
||||
"url",
|
||||
"async",
|
||||
"interrupt_on_error",
|
||||
"count",
|
||||
}
|
||||
|
||||
prepareTargetStmt = `SELECT projections.targets.id,` +
|
||||
` projections.targets.change_date,` +
|
||||
` projections.targets.resource_owner,` +
|
||||
` projections.targets.sequence,` +
|
||||
` projections.targets.name,` +
|
||||
` projections.targets.target_type,` +
|
||||
` projections.targets.timeout,` +
|
||||
` projections.targets.url,` +
|
||||
` projections.targets.async,` +
|
||||
` projections.targets.interrupt_on_error` +
|
||||
` FROM projections.targets`
|
||||
prepareTargetCols = []string{
|
||||
"id",
|
||||
"change_date",
|
||||
"resource_owner",
|
||||
"sequence",
|
||||
"name",
|
||||
"target_type",
|
||||
"timeout",
|
||||
"url",
|
||||
"async",
|
||||
"interrupt_on_error",
|
||||
}
|
||||
)
|
||||
|
||||
func Test_TargetPrepares(t *testing.T) {
|
||||
type want struct {
|
||||
sqlExpectations sqlExpectation
|
||||
err checkErr
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
prepare interface{}
|
||||
want want
|
||||
object interface{}
|
||||
}{
|
||||
{
|
||||
name: "prepareTargetsQuery no result",
|
||||
prepare: prepareTargetsQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueries(
|
||||
regexp.QuoteMeta(prepareTargetsStmt),
|
||||
nil,
|
||||
nil,
|
||||
),
|
||||
},
|
||||
object: &Targets{Targets: []*Target{}},
|
||||
},
|
||||
{
|
||||
name: "prepareTargetsQuery one result",
|
||||
prepare: prepareTargetsQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueries(
|
||||
regexp.QuoteMeta(prepareTargetsStmt),
|
||||
prepareTargetsCols,
|
||||
[][]driver.Value{
|
||||
{
|
||||
"id",
|
||||
testNow,
|
||||
"ro",
|
||||
uint64(20211109),
|
||||
"target-name",
|
||||
domain.TargetTypeWebhook,
|
||||
1 * time.Second,
|
||||
"https://example.com",
|
||||
true,
|
||||
true,
|
||||
},
|
||||
},
|
||||
),
|
||||
},
|
||||
object: &Targets{
|
||||
SearchResponse: SearchResponse{
|
||||
Count: 1,
|
||||
},
|
||||
Targets: []*Target{
|
||||
{
|
||||
ID: "id",
|
||||
ObjectDetails: domain.ObjectDetails{
|
||||
EventDate: testNow,
|
||||
ResourceOwner: "ro",
|
||||
Sequence: 20211109,
|
||||
},
|
||||
Name: "target-name",
|
||||
TargetType: domain.TargetTypeWebhook,
|
||||
Timeout: 1 * time.Second,
|
||||
URL: "https://example.com",
|
||||
Async: true,
|
||||
InterruptOnError: true,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "prepareTargetsQuery multiple result",
|
||||
prepare: prepareTargetsQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueries(
|
||||
regexp.QuoteMeta(prepareTargetsStmt),
|
||||
prepareTargetsCols,
|
||||
[][]driver.Value{
|
||||
{
|
||||
"id-1",
|
||||
testNow,
|
||||
"ro",
|
||||
uint64(20211109),
|
||||
"target-name1",
|
||||
domain.TargetTypeWebhook,
|
||||
1 * time.Second,
|
||||
"https://example.com",
|
||||
true,
|
||||
false,
|
||||
},
|
||||
{
|
||||
"id-2",
|
||||
testNow,
|
||||
"ro",
|
||||
uint64(20211110),
|
||||
"target-name2",
|
||||
domain.TargetTypeWebhook,
|
||||
1 * time.Second,
|
||||
"https://example.com",
|
||||
false,
|
||||
true,
|
||||
},
|
||||
},
|
||||
),
|
||||
},
|
||||
object: &Targets{
|
||||
SearchResponse: SearchResponse{
|
||||
Count: 2,
|
||||
},
|
||||
Targets: []*Target{
|
||||
{
|
||||
ID: "id-1",
|
||||
ObjectDetails: domain.ObjectDetails{
|
||||
EventDate: testNow,
|
||||
ResourceOwner: "ro",
|
||||
Sequence: 20211109,
|
||||
},
|
||||
Name: "target-name1",
|
||||
TargetType: domain.TargetTypeWebhook,
|
||||
Timeout: 1 * time.Second,
|
||||
URL: "https://example.com",
|
||||
Async: true,
|
||||
InterruptOnError: false,
|
||||
},
|
||||
{
|
||||
ID: "id-2",
|
||||
ObjectDetails: domain.ObjectDetails{
|
||||
EventDate: testNow,
|
||||
ResourceOwner: "ro",
|
||||
Sequence: 20211110,
|
||||
},
|
||||
Name: "target-name2",
|
||||
TargetType: domain.TargetTypeWebhook,
|
||||
Timeout: 1 * time.Second,
|
||||
URL: "https://example.com",
|
||||
Async: false,
|
||||
InterruptOnError: true,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "prepareTargetsQuery sql err",
|
||||
prepare: prepareTargetsQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueryErr(
|
||||
regexp.QuoteMeta(prepareTargetsStmt),
|
||||
sql.ErrConnDone,
|
||||
),
|
||||
err: func(err error) (error, bool) {
|
||||
if !errors.Is(err, sql.ErrConnDone) {
|
||||
return fmt.Errorf("err should be sql.ErrConnDone got: %w", err), false
|
||||
}
|
||||
return nil, true
|
||||
},
|
||||
},
|
||||
object: (*Target)(nil),
|
||||
},
|
||||
{
|
||||
name: "prepareTargetQuery no result",
|
||||
prepare: prepareTargetQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueriesScanErr(
|
||||
regexp.QuoteMeta(prepareTargetStmt),
|
||||
nil,
|
||||
nil,
|
||||
),
|
||||
err: func(err error) (error, bool) {
|
||||
if !zerrors.IsNotFound(err) {
|
||||
return fmt.Errorf("err should be zitadel.NotFoundError got: %w", err), false
|
||||
}
|
||||
return nil, true
|
||||
},
|
||||
},
|
||||
object: (*Target)(nil),
|
||||
},
|
||||
{
|
||||
name: "prepareTargetQuery found",
|
||||
prepare: prepareTargetQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQuery(
|
||||
regexp.QuoteMeta(prepareTargetStmt),
|
||||
prepareTargetCols,
|
||||
[]driver.Value{
|
||||
"id",
|
||||
testNow,
|
||||
"ro",
|
||||
uint64(20211109),
|
||||
"target-name",
|
||||
domain.TargetTypeWebhook,
|
||||
1 * time.Second,
|
||||
"https://example.com",
|
||||
true,
|
||||
false,
|
||||
},
|
||||
),
|
||||
},
|
||||
object: &Target{
|
||||
ID: "id",
|
||||
ObjectDetails: domain.ObjectDetails{
|
||||
EventDate: testNow,
|
||||
ResourceOwner: "ro",
|
||||
Sequence: 20211109,
|
||||
},
|
||||
Name: "target-name",
|
||||
TargetType: domain.TargetTypeWebhook,
|
||||
Timeout: 1 * time.Second,
|
||||
URL: "https://example.com",
|
||||
Async: true,
|
||||
InterruptOnError: false,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "prepareTargetQuery sql err",
|
||||
prepare: prepareTargetQuery,
|
||||
want: want{
|
||||
sqlExpectations: mockQueryErr(
|
||||
regexp.QuoteMeta(prepareTargetStmt),
|
||||
sql.ErrConnDone,
|
||||
),
|
||||
err: func(err error) (error, bool) {
|
||||
if !errors.Is(err, sql.ErrConnDone) {
|
||||
return fmt.Errorf("err should be sql.ErrConnDone got: %w", err), false
|
||||
}
|
||||
return nil, true
|
||||
},
|
||||
},
|
||||
object: (*Target)(nil),
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
assertPrepare(t, tt.prepare, tt.object, tt.want.sqlExpectations, tt.want.err, defaultPrepareArgs...)
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user