mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-16 04:48:04 +00:00
b522588d98
# Which Problems Are Solved Float64 which was used for the event.Position field is [not precise in go and gets rounded](https://github.com/golang/go/issues/47300). This can lead to unprecies position tracking of events and therefore projections especially on cockcoachdb as the position used there is a big number. example of a unprecies position: exact: 1725257931223002628 float64: 1725257931223002624.000000 # How the Problems Are Solved The float64 was replaced by [github.com/jackc/pgx-shopspring-decimal](https://github.com/jackc/pgx-shopspring-decimal). # Additional Changes Correct behaviour of makefile for load tests. Rename `latestSequence`-queries to `latestPosition`
99 lines
2.5 KiB
Go
99 lines
2.5 KiB
Go
package mirror
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/shopspring/decimal"
|
|
|
|
"github.com/zitadel/zitadel/internal/v2/eventstore"
|
|
"github.com/zitadel/zitadel/internal/v2/projection"
|
|
"github.com/zitadel/zitadel/internal/v2/readmodel"
|
|
"github.com/zitadel/zitadel/internal/v2/system"
|
|
mirror_event "github.com/zitadel/zitadel/internal/v2/system/mirror"
|
|
)
|
|
|
|
func queryLastSuccessfulMigration(ctx context.Context, destinationES *eventstore.EventStore, source string) (*readmodel.LastSuccessfulMirror, error) {
|
|
lastSuccess := readmodel.NewLastSuccessfulMirror(source)
|
|
if shouldIgnorePrevious {
|
|
return lastSuccess, nil
|
|
}
|
|
_, err := destinationES.Query(
|
|
ctx,
|
|
eventstore.NewQuery(
|
|
system.AggregateInstance,
|
|
lastSuccess,
|
|
eventstore.SetFilters(lastSuccess.Filter()),
|
|
),
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return lastSuccess, nil
|
|
}
|
|
|
|
func writeMigrationStart(ctx context.Context, sourceES *eventstore.EventStore, id string, destination string) (_ decimal.Decimal, err error) {
|
|
var cmd *eventstore.Command
|
|
if len(instanceIDs) > 0 {
|
|
cmd, err = mirror_event.NewStartedInstancesCommand(destination, instanceIDs)
|
|
if err != nil {
|
|
return decimal.Decimal{}, err
|
|
}
|
|
} else {
|
|
cmd = mirror_event.NewStartedSystemCommand(destination)
|
|
}
|
|
|
|
var position projection.HighestPosition
|
|
|
|
err = sourceES.Push(
|
|
ctx,
|
|
eventstore.NewPushIntent(
|
|
system.AggregateInstance,
|
|
eventstore.AppendAggregate(
|
|
system.AggregateOwner,
|
|
system.AggregateType,
|
|
id,
|
|
eventstore.CurrentSequenceMatches(0),
|
|
eventstore.AppendCommands(cmd),
|
|
),
|
|
eventstore.PushReducer(&position),
|
|
),
|
|
)
|
|
if err != nil {
|
|
return decimal.Decimal{}, err
|
|
}
|
|
return position.Position, nil
|
|
}
|
|
|
|
func writeMigrationSucceeded(ctx context.Context, destinationES *eventstore.EventStore, id, source string, position decimal.Decimal) error {
|
|
return destinationES.Push(
|
|
ctx,
|
|
eventstore.NewPushIntent(
|
|
system.AggregateInstance,
|
|
eventstore.AppendAggregate(
|
|
system.AggregateOwner,
|
|
system.AggregateType,
|
|
id,
|
|
eventstore.CurrentSequenceMatches(0),
|
|
eventstore.AppendCommands(mirror_event.NewSucceededCommand(source, position)),
|
|
),
|
|
),
|
|
)
|
|
}
|
|
|
|
func writeMigrationFailed(ctx context.Context, destinationES *eventstore.EventStore, id, source string, err error) error {
|
|
return destinationES.Push(
|
|
ctx,
|
|
eventstore.NewPushIntent(
|
|
system.AggregateInstance,
|
|
eventstore.AppendAggregate(
|
|
system.AggregateOwner,
|
|
system.AggregateType,
|
|
id,
|
|
eventstore.CurrentSequenceMatches(0),
|
|
eventstore.AppendCommands(mirror_event.NewFailedCommand(source, err)),
|
|
),
|
|
),
|
|
)
|
|
}
|