TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-08-14 08:57:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert "fix(eventstore): use decimal for position (#9881)"

Browse Source 
This reverts commit e14639c0ad.
This commit is contained in:
adlerhurst
2025-05-19 17:14:47 +02:00
parent 3883ffd1ce
commit 90f7310848
49 changed files with 236 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
internal/eventstore/handler/v2/field_handler.go
View File

@@ -8,7 +8,6 @@ import (
"time"
"github.com/jackc/pgx/v5/pgconn"
"github.com/shopspring/decimal"
"github.com/zitadel/zitadel/internal/eventstore"
)
@@ -127,15 +126,10 @@ func (h *FieldHandler) processEvents(ctx context.Context, config *triggerConfig)
return additionalIteration, err
}
// stop execution if currentState.eventTimestamp >= config.maxCreatedAt
if !config.maxPosition.IsZero() && currentState.position.GreaterThanOrEqual(config.maxPosition) {
if config.maxPosition != 0 && currentState.position >= config.maxPosition {
return false, nil
}
if config.minPosition.GreaterThan(decimal.NewFromInt(0)) {
currentState.position = config.minPosition
currentState.offset = 0
}
events, additionalIteration, err := h.fetchEvents(ctx, tx, currentState)
if err != nil {
return additionalIteration, err
@@ -165,7 +159,7 @@ func (h *FieldHandler) fetchEvents(ctx context.Context, tx *sql.Tx, currentState
idx, offset := skipPreviouslyReducedEvents(events, currentState)
if currentState.position.Equal(events[len(events)-1].Position()) {
if currentState.position == events[len(events)-1].Position() {
offset += currentState.offset
}
currentState.position = events[len(events)-1].Position()
@@ -195,9 +189,9 @@ func (h *FieldHandler) fetchEvents(ctx context.Context, tx *sql.Tx, currentState
}
func skipPreviouslyReducedEvents(events []eventstore.Event, currentState *state) (index int, offset uint32) {
var position decimal.Decimal
var position float64
for i, event := range events {
if !event.Position().Equal(position) {
if event.Position() != position {
offset = 0
position = event.Position()
}

27
internal/eventstore/handler/v2/handler.go
View File

@@ -4,13 +4,13 @@ import (
"context"
"database/sql"
"errors"
"math"
"math/rand"
"slices"
"sync"
"time"
"github.com/jackc/pgx/v5/pgconn"
"github.com/shopspring/decimal"
"github.com/zitadel/logging"
"github.com/zitadel/zitadel/internal/api/authz"
@@ -381,8 +381,7 @@ func (h *Handler) existingInstances(ctx context.Context) ([]string, error) {
type triggerConfig struct {
awaitRunning bool
maxPosition decimal.Decimal
minPosition decimal.Decimal
maxPosition float64
}
type TriggerOpt func(conf *triggerConfig)
@@ -393,18 +392,12 @@ func WithAwaitRunning() TriggerOpt {
}
}
func WithMaxPosition(position decimal.Decimal) TriggerOpt {
func WithMaxPosition(position float64) TriggerOpt {
return func(conf *triggerConfig) {
conf.maxPosition = position
}
}
func WithMinPosition(position decimal.Decimal) TriggerOpt {
return func(conf *triggerConfig) {
conf.minPosition = position
}
}
func (h *Handler) Trigger(ctx context.Context, opts ...TriggerOpt) (_ context.Context, err error) {
config := new(triggerConfig)
for _, opt := range opts {
@@ -513,15 +506,10 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add
return additionalIteration, err
}
// stop execution if currentState.position >= config.maxPosition
if !config.maxPosition.Equal(decimal.Decimal{}) && currentState.position.GreaterThanOrEqual(config.maxPosition) {
if config.maxPosition != 0 && currentState.position >= config.maxPosition {
return false, nil
}
if config.minPosition.GreaterThan(decimal.NewFromInt(0)) {
currentState.position = config.minPosition
currentState.offset = 0
}
var statements []*Statement
statements, additionalIteration, err = h.generateStatements(ctx, tx, currentState)
if err != nil {
@@ -613,7 +601,7 @@ func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentSta
func skipPreviouslyReducedStatements(statements []*Statement, currentState *state) int {
for i, statement := range statements {
if statement.Position.Equal(currentState.position) &&
if statement.Position == currentState.position &&
statement.Aggregate.ID == currentState.aggregateID &&
statement.Aggregate.Type == currentState.aggregateType &&
statement.Sequence == currentState.sequence {
@@ -677,8 +665,9 @@ func (h *Handler) eventQuery(currentState *state) *eventstore.SearchQueryBuilder
OrderAsc().
InstanceID(currentState.instanceID)
if currentState.position.GreaterThan(decimal.Decimal{}) {
builder = builder.PositionAtLeast(currentState.position)
if currentState.position > 0 {
// decrease position by 10 because builder.PositionAfter filters for position > and we need position >=
builder = builder.PositionAfter(math.Float64frombits(math.Float64bits(currentState.position) - 10))
if currentState.offset > 0 {
builder = builder.Offset(currentState.offset)
}

8
internal/eventstore/handler/v2/state.go
View File

@@ -7,8 +7,6 @@ import (
"errors"
"time"
"github.com/shopspring/decimal"
"github.com/zitadel/zitadel/internal/api/authz"
"github.com/zitadel/zitadel/internal/eventstore"
"github.com/zitadel/zitadel/internal/zerrors"
@@ -16,7 +14,7 @@ import (
type state struct {
instanceID string
position decimal.Decimal
position float64
eventTimestamp time.Time
aggregateType eventstore.AggregateType
aggregateID string
@@ -47,7 +45,7 @@ func (h *Handler) currentState(ctx context.Context, tx *sql.Tx, config *triggerC
aggregateType = new(sql.NullString)
sequence = new(sql.NullInt64)
timestamp = new(sql.NullTime)
position = new(decimal.NullDecimal)
position = new(sql.NullFloat64)
offset = new(sql.NullInt64)
)
@@ -77,7 +75,7 @@ func (h *Handler) currentState(ctx context.Context, tx *sql.Tx, config *triggerC
currentState.aggregateType = eventstore.AggregateType(aggregateType.String)
currentState.sequence = uint64(sequence.Int64)
currentState.eventTimestamp = timestamp.Time
currentState.position = position.Decimal
currentState.position = position.Float64
// psql does not provide unsigned numbers so we work around it
currentState.offset = uint32(offset.Int64)
return currentState, nil

13
internal/eventstore/handler/v2/state_test.go
View File

@@ -11,7 +11,6 @@ import (
"time"
"github.com/jackc/pgx/v5/pgconn"
"github.com/shopspring/decimal"
"github.com/zitadel/zitadel/internal/api/authz"
"github.com/zitadel/zitadel/internal/database/mock"
@@ -167,7 +166,7 @@ func TestHandler_updateLastUpdated(t *testing.T) {
updatedState: &state{
instanceID: "instance",
eventTimestamp: time.Now(),
position: decimal.NewFromInt(42),
position: 42,
},
},
isErr: func(t *testing.T, err error) {
@@ -193,7 +192,7 @@ func TestHandler_updateLastUpdated(t *testing.T) {
updatedState: &state{
instanceID: "instance",
eventTimestamp: time.Now(),
position: decimal.NewFromInt(42),
position: 42,
},
},
isErr: func(t *testing.T, err error) {
@@ -218,7 +217,7 @@ func TestHandler_updateLastUpdated(t *testing.T) {
eventstore.AggregateType("aggregate type"),
uint64(42),
mock.AnyType[time.Time]{},
decimal.NewFromInt(42),
float64(42),
uint32(0),
),
mock.WithExecRowsAffected(1),
@@ -229,7 +228,7 @@ func TestHandler_updateLastUpdated(t *testing.T) {
updatedState: &state{
instanceID: "instance",
eventTimestamp: time.Now(),
position: decimal.NewFromInt(42),
position: 42,
aggregateType: "aggregate type",
aggregateID: "aggregate id",
sequence: 42,
@@ -398,7 +397,7 @@ func TestHandler_currentState(t *testing.T) {
"aggregate type",
int64(42),
testTime,
decimal.NewFromInt(42).String(),
float64(42),
uint16(10),
},
},
@@ -413,7 +412,7 @@ func TestHandler_currentState(t *testing.T) {
currentState: &state{
instanceID: "instance",
eventTimestamp: testTime,
position: decimal.NewFromInt(42),
position: 42,
aggregateType: "aggregate type",
aggregateID: "aggregate id",
sequence: 42,

3
internal/eventstore/handler/v2/statement.go
View File

@@ -9,7 +9,6 @@ import (
"strings"
"time"
"github.com/shopspring/decimal"
"github.com/zitadel/logging"
"golang.org/x/exp/constraints"
@@ -83,7 +82,7 @@ func (h *Handler) reduce(event eventstore.Event) (*Statement, error) {
type Statement struct {
Aggregate *eventstore.Aggregate
Sequence uint64
Position decimal.Decimal
Position float64
CreationDate time.Time
offset uint32