perf: query data AS OF SYSTEM TIME (#5231)

Queries the data in the storage layser at the timestamp when the call hit the API layer

internal/eventstore/handler/crdb/handler_stmt.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/zitadel/logging"
 
+	"github.com/zitadel/zitadel/internal/database"
 	"github.com/zitadel/zitadel/internal/errors"
 	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/eventstore/handler"
@@ -19,7 +20,7 @@ var (
 type StatementHandlerConfig struct {
 	handler.ProjectionHandlerConfig
 
-	Client            *sql.DB
+	Client            *database.DB
 	SequenceTable     string
 	LockTable         string
 	FailedEventsTable string
@@ -34,7 +35,7 @@ type StatementHandler struct {
 	*handler.ProjectionHandler
 	Locker
 
-	client                  *sql.DB
+	client                  *database.DB
 	sequenceTable           string
 	currentSequenceStmt     string
 	updateSequencesBaseStmt string
@@ -74,7 +75,7 @@ func NewStatementHandler(
 		aggregates:              aggregateTypes,
 		reduces:                 reduces,
 		bulkLimit:               config.BulkLimit,
-		Locker:                  NewLocker(config.Client, config.LockTable, config.ProjectionName),
+		Locker:                  NewLocker(config.Client.DB, config.LockTable, config.ProjectionName),
 		initCheck:               config.InitCheck,
 		initialized:             make(chan bool),
 	}
@@ -96,7 +97,7 @@ func (h *StatementHandler) SearchQuery(ctx context.Context, instanceIDs []string
 		return nil, 0, err
 	}
 
-	queryBuilder := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).Limit(h.bulkLimit)
+	queryBuilder := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).Limit(h.bulkLimit).AllowTimeTravel()
 
 	for _, aggregateType := range h.aggregates {
 		for _, instanceID := range instanceIDs {

internal/eventstore/handler/crdb/handler_stmt_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/zitadel/zitadel/internal/database"
 	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/eventstore/handler"
 	"github.com/zitadel/zitadel/internal/eventstore/repository"
@@ -114,6 +115,7 @@ func TestProjectionHandler_SearchQuery(t *testing.T) {
 				},
 				SearchQueryBuilder: eventstore.
 					NewSearchQueryBuilder(eventstore.ColumnsEvent).
+					AllowTimeTravel().
 					AddQuery().
 					AggregateTypes("testAgg").
 					SequenceGreater(5).
@@ -143,6 +145,7 @@ func TestProjectionHandler_SearchQuery(t *testing.T) {
 				},
 				SearchQueryBuilder: eventstore.
 					NewSearchQueryBuilder(eventstore.ColumnsEvent).
+					AllowTimeTravel().
 					AddQuery().
 					AggregateTypes("testAgg").
 					SequenceGreater(5).
@@ -171,7 +174,9 @@ func TestProjectionHandler_SearchQuery(t *testing.T) {
 				},
 				SequenceTable: tt.fields.sequenceTable,
 				BulkLimit:     tt.fields.bulkLimit,
-				Client:        client,
+				Client: &database.DB{
+					DB: client,
+				},
 			})
 
 			h.aggregates = tt.fields.aggregates
@@ -549,7 +554,9 @@ func TestStatementHandler_Update(t *testing.T) {
 				sequenceTable:           "my_sequences",
 				currentSequenceStmt:     fmt.Sprintf(currentSequenceStmtFormat, "my_sequences"),
 				updateSequencesBaseStmt: fmt.Sprintf(updateCurrentSequencesStmtFormat, "my_sequences"),
-				client:                  client,
+				client: &database.DB{
+					DB: client,
+				},
 			}
 
 			h.aggregates = tt.fields.aggregates
@@ -1121,7 +1128,9 @@ func TestStatementHandler_executeStmts(t *testing.T) {
 						ProjectionName: tt.fields.projectionName,
 						RequeueEvery:   0,
 					},
-					Client:            client,
+					Client: &database.DB{
+						DB: client,
+					},
 					FailedEventsTable: tt.fields.failedEventsTable,
 					MaxFailureCount:   tt.fields.maxFailureCount,
 				},

internal/eventstore/handler/handler_projection.go

@@ -194,7 +194,7 @@ func (h *ProjectionHandler) schedule(ctx context.Context) {
 	var succeededOnce bool
 	var err error
 	// get every instance id except empty (system)
-	query := eventstore.NewSearchQueryBuilder(eventstore.ColumnsInstanceIDs).AddQuery().ExcludedInstanceID("")
+	query := eventstore.NewSearchQueryBuilder(eventstore.ColumnsInstanceIDs).AllowTimeTravel().AddQuery().ExcludedInstanceID("")
 	for range h.triggerProjection.C {
 		if !succeededOnce {
 			// (re)check if it has succeeded in the meantime