mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-16 12:58:00 +00:00
326 lines
8.3 KiB
Go
326 lines
8.3 KiB
Go
package sql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/zitadel/logging"
|
|
|
|
"github.com/zitadel/zitadel/internal/api/call"
|
|
"github.com/zitadel/zitadel/internal/database"
|
|
"github.com/zitadel/zitadel/internal/database/dialect"
|
|
"github.com/zitadel/zitadel/internal/eventstore"
|
|
"github.com/zitadel/zitadel/internal/eventstore/repository"
|
|
"github.com/zitadel/zitadel/internal/zerrors"
|
|
)
|
|
|
|
type querier interface {
|
|
columnName(field repository.Field, useV1 bool) string
|
|
operation(repository.Operation) string
|
|
conditionFormat(repository.Operation) string
|
|
placeholder(query string) string
|
|
eventQuery(useV1 bool) string
|
|
maxSequenceQuery(useV1 bool) string
|
|
instanceIDsQuery(useV1 bool) string
|
|
db() *database.DB
|
|
orderByEventSequence(desc, shouldOrderBySequence, useV1 bool) string
|
|
dialect.Database
|
|
}
|
|
|
|
type scan func(dest ...interface{}) error
|
|
|
|
type tx struct {
|
|
*sql.Tx
|
|
}
|
|
|
|
func (t *tx) QueryContext(ctx context.Context, scan func(rows *sql.Rows) error, query string, args ...any) error {
|
|
rows, err := t.Tx.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() {
|
|
closeErr := rows.Close()
|
|
logging.OnError(closeErr).Info("rows.Close failed")
|
|
}()
|
|
|
|
if err = scan(rows); err != nil {
|
|
return err
|
|
}
|
|
return rows.Err()
|
|
}
|
|
|
|
func query(ctx context.Context, criteria querier, searchQuery *eventstore.SearchQueryBuilder, dest interface{}, useV1 bool) error {
|
|
q, err := repository.QueryFromBuilder(searchQuery)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
query, rowScanner := prepareColumns(criteria, q.Columns, useV1)
|
|
where, values := prepareConditions(criteria, q, useV1)
|
|
if where == "" || query == "" {
|
|
return zerrors.ThrowInvalidArgument(nil, "SQL-rWeBw", "invalid query factory")
|
|
}
|
|
if q.Tx == nil {
|
|
if travel := prepareTimeTravel(ctx, criteria, q.AllowTimeTravel); travel != "" {
|
|
query += travel
|
|
}
|
|
}
|
|
query += where
|
|
|
|
// instead of using the max function of the database (which doesn't work for postgres)
|
|
// we select the most recent row
|
|
if q.Columns == eventstore.ColumnsMaxSequence {
|
|
q.Limit = 1
|
|
q.Desc = true
|
|
}
|
|
|
|
// if there is only one subquery we can optimize the query ordering by ordering by sequence
|
|
var shouldOrderBySequence bool
|
|
if len(q.SubQueries) == 1 {
|
|
for _, filter := range q.SubQueries[0] {
|
|
if filter.Field == repository.FieldAggregateID {
|
|
shouldOrderBySequence = filter.Operation == repository.OperationEquals
|
|
}
|
|
}
|
|
}
|
|
|
|
switch q.Columns {
|
|
case eventstore.ColumnsEvent,
|
|
eventstore.ColumnsMaxSequence:
|
|
query += criteria.orderByEventSequence(q.Desc, shouldOrderBySequence, useV1)
|
|
}
|
|
|
|
if q.Limit > 0 {
|
|
values = append(values, q.Limit)
|
|
query += " LIMIT ?"
|
|
}
|
|
|
|
if q.Offset > 0 {
|
|
values = append(values, q.Offset)
|
|
query += " OFFSET ?"
|
|
}
|
|
|
|
query = criteria.placeholder(query)
|
|
|
|
var contextQuerier interface {
|
|
QueryContext(context.Context, func(rows *sql.Rows) error, string, ...interface{}) error
|
|
}
|
|
contextQuerier = criteria.db()
|
|
if q.Tx != nil {
|
|
contextQuerier = &tx{Tx: q.Tx}
|
|
}
|
|
|
|
err = contextQuerier.QueryContext(ctx,
|
|
func(rows *sql.Rows) error {
|
|
for rows.Next() {
|
|
err := rowScanner(rows.Scan, dest)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}, query, values...)
|
|
if err != nil {
|
|
logging.New().WithError(err).Info("query failed")
|
|
return zerrors.ThrowInternal(err, "SQL-KyeAx", "unable to filter events")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func prepareColumns(criteria querier, columns eventstore.Columns, useV1 bool) (string, func(s scan, dest interface{}) error) {
|
|
switch columns {
|
|
case eventstore.ColumnsMaxSequence:
|
|
return criteria.maxSequenceQuery(useV1), maxSequenceScanner
|
|
case eventstore.ColumnsInstanceIDs:
|
|
return criteria.instanceIDsQuery(useV1), instanceIDsScanner
|
|
case eventstore.ColumnsEvent:
|
|
return criteria.eventQuery(useV1), eventsScanner(useV1)
|
|
default:
|
|
return "", nil
|
|
}
|
|
}
|
|
|
|
func prepareTimeTravel(ctx context.Context, criteria querier, allow bool) string {
|
|
if !allow {
|
|
return ""
|
|
}
|
|
took := call.Took(ctx)
|
|
return criteria.Timetravel(took)
|
|
}
|
|
|
|
func maxSequenceScanner(row scan, dest interface{}) (err error) {
|
|
position, ok := dest.(*sql.NullFloat64)
|
|
if !ok {
|
|
return zerrors.ThrowInvalidArgumentf(nil, "SQL-NBjA9", "type must be sql.NullInt64 got: %T", dest)
|
|
}
|
|
err = row(position)
|
|
if err == nil || errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
}
|
|
return zerrors.ThrowInternal(err, "SQL-bN5xg", "something went wrong")
|
|
}
|
|
|
|
func instanceIDsScanner(scanner scan, dest interface{}) (err error) {
|
|
ids, ok := dest.(*[]string)
|
|
if !ok {
|
|
return zerrors.ThrowInvalidArgument(nil, "SQL-Begh2", "type must be an array of string")
|
|
}
|
|
var id string
|
|
err = scanner(&id)
|
|
if err != nil {
|
|
logging.WithError(err).Warn("unable to scan row")
|
|
return zerrors.ThrowInternal(err, "SQL-DEFGe", "unable to scan row")
|
|
}
|
|
*ids = append(*ids, id)
|
|
|
|
return nil
|
|
}
|
|
|
|
func eventsScanner(useV1 bool) func(scanner scan, dest interface{}) (err error) {
|
|
return func(scanner scan, dest interface{}) (err error) {
|
|
reduce, ok := dest.(eventstore.Reducer)
|
|
if !ok {
|
|
return zerrors.ThrowInvalidArgumentf(nil, "SQL-4GP6F", "events scanner: invalid type %T", dest)
|
|
}
|
|
event := new(repository.Event)
|
|
position := new(sql.NullFloat64)
|
|
|
|
if useV1 {
|
|
err = scanner(
|
|
&event.CreationDate,
|
|
&event.Typ,
|
|
&event.Seq,
|
|
&event.Data,
|
|
&event.EditorUser,
|
|
&event.ResourceOwner,
|
|
&event.InstanceID,
|
|
&event.AggregateType,
|
|
&event.AggregateID,
|
|
&event.Version,
|
|
)
|
|
} else {
|
|
var revision uint8
|
|
err = scanner(
|
|
&event.CreationDate,
|
|
&event.Typ,
|
|
&event.Seq,
|
|
position,
|
|
&event.Data,
|
|
&event.EditorUser,
|
|
&event.ResourceOwner,
|
|
&event.InstanceID,
|
|
&event.AggregateType,
|
|
&event.AggregateID,
|
|
&revision,
|
|
)
|
|
event.Version = eventstore.Version("v" + strconv.Itoa(int(revision)))
|
|
}
|
|
|
|
if err != nil {
|
|
logging.New().WithError(err).Warn("unable to scan row")
|
|
return zerrors.ThrowInternal(err, "SQL-M0dsf", "unable to scan row")
|
|
}
|
|
event.Pos = position.Float64
|
|
return reduce(event)
|
|
}
|
|
}
|
|
|
|
func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bool) (_ string, args []any) {
|
|
clauses, args := prepareQuery(criteria, useV1, query.InstanceID, query.InstanceIDs, query.ExcludedInstances)
|
|
if clauses != "" && len(query.SubQueries) > 0 {
|
|
clauses += " AND "
|
|
}
|
|
subClauses := make([]string, len(query.SubQueries))
|
|
for i, filters := range query.SubQueries {
|
|
var subArgs []any
|
|
subClauses[i], subArgs = prepareQuery(criteria, useV1, filters...)
|
|
// an error is thrown in [query]
|
|
if subClauses[i] == "" {
|
|
return "", nil
|
|
}
|
|
if len(query.SubQueries) > 1 && len(subArgs) > 1 {
|
|
subClauses[i] = "(" + subClauses[i] + ")"
|
|
}
|
|
args = append(args, subArgs...)
|
|
}
|
|
if len(subClauses) == 1 {
|
|
clauses += subClauses[0]
|
|
} else if len(subClauses) > 1 {
|
|
clauses += "(" + strings.Join(subClauses, " OR ") + ")"
|
|
}
|
|
|
|
additionalClauses, additionalArgs := prepareQuery(criteria, useV1,
|
|
query.Position,
|
|
query.Owner,
|
|
query.Sequence,
|
|
query.CreatedAfter,
|
|
query.CreatedBefore,
|
|
query.Creator,
|
|
)
|
|
if additionalClauses != "" {
|
|
if clauses != "" {
|
|
clauses += " AND "
|
|
}
|
|
clauses += additionalClauses
|
|
args = append(args, additionalArgs...)
|
|
}
|
|
|
|
if query.AwaitOpenTransactions {
|
|
clauses += awaitOpenTransactions(useV1)
|
|
}
|
|
|
|
if clauses == "" {
|
|
return "", nil
|
|
}
|
|
|
|
return " WHERE " + clauses, args
|
|
}
|
|
|
|
func prepareQuery(criteria querier, useV1 bool, filters ...*repository.Filter) (_ string, args []any) {
|
|
clauses := make([]string, 0, len(filters))
|
|
args = make([]any, 0, len(filters))
|
|
for _, filter := range filters {
|
|
if filter == nil {
|
|
continue
|
|
}
|
|
arg := filter.Value
|
|
|
|
// marshal if payload filter
|
|
if filter.Field == repository.FieldEventData {
|
|
var err error
|
|
arg, err = json.Marshal(arg)
|
|
if err != nil {
|
|
logging.WithError(err).Warn("unable to marshal search value")
|
|
continue
|
|
}
|
|
|
|
}
|
|
|
|
clauses = append(clauses, getCondition(criteria, filter, useV1))
|
|
// if mapping failed an error is thrown in [query]
|
|
if clauses[len(clauses)-1] == "" {
|
|
return "", nil
|
|
}
|
|
args = append(args, arg)
|
|
}
|
|
|
|
return strings.Join(clauses, " AND "), args
|
|
}
|
|
|
|
func getCondition(cond querier, filter *repository.Filter, useV1 bool) (condition string) {
|
|
field := cond.columnName(filter.Field, useV1)
|
|
operation := cond.operation(filter.Operation)
|
|
if field == "" || operation == "" {
|
|
return ""
|
|
}
|
|
format := cond.conditionFormat(filter.Operation)
|
|
|
|
return fmt.Sprintf(format, field, operation)
|
|
}
|