TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-08-13 10:07:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(eventstore): use decimal, correct mirror (#9905)

Browse Source 
back port #9812, #9878, #9881, #9884

---------

Co-authored-by: Livio Spring <livio.a@gmail.com>
Co-authored-by: Stefan Benz <46600784+stebenz@users.noreply.github.com>
This commit is contained in:
Silvan
2025-05-20 12:58:12 +02:00
committed by GitHub
parent 1df6638847
commit f3338cdb9a
58 changed files with 545 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
cmd/mirror/projections.go
View File

@@ -3,6 +3,7 @@ package mirror
import (
"context"
"database/sql"
"fmt"
"net/http"
"sync"
"time"
@@ -103,6 +104,7 @@ func projections(
config *ProjectionsConfig,
masterKey string,
) {
logging.Info("starting to fill projections")
start := time.Now()
client, err := database.Connect(config.Destination, false)
@@ -118,7 +120,11 @@ func projections(
logging.OnError(err).Fatal("unable create static storage")
config.Eventstore.Querier = old_es.NewCRDB(client)
config.Eventstore.Pusher = new_es.NewEventstore(client)
newES := new_es.NewEventstore(client)
config.Eventstore.Pusher = newES
config.Eventstore.Searcher = newES
es := eventstore.NewEventstore(config.Eventstore)
esV4 := es_v4.NewEventstoreFromOne(es_v4_pg.New(client, &es_v4_pg.Config{
MaxRetries: config.Eventstore.MaxRetries,
@@ -251,8 +257,10 @@ func projections(
go execProjections(ctx, instances, failedInstances, &wg)
}
for _, instance := range queryInstanceIDs(ctx, client) {
existingInstances := queryInstanceIDs(ctx, client)
for i, instance := range existingInstances {
instances <- instance
logging.WithFields("id", instance, "index", fmt.Sprintf("%d/%d", i, len(existingInstances))).Info("instance queued for projection")
}
close(instances)
wg.Wait()
@@ -264,7 +272,7 @@ func projections(
func execProjections(ctx context.Context, instances <-chan string, failedInstances chan<- string, wg *sync.WaitGroup) {
for instance := range instances {
logging.WithFields("instance", instance).Info("start projections")
logging.WithFields("instance", instance).Info("starting projections")
ctx = internal_authz.WithInstanceID(ctx, instance)
err := projection.ProjectInstance(ctx)
@@ -281,6 +289,13 @@ func execProjections(ctx context.Context, instances <-chan string, failedInstanc
continue
}
err = projection.ProjectInstanceFields(ctx)
if err != nil {
logging.WithFields("instance", instance).WithError(err).Info("trigger fields failed")
failedInstances <- instance
continue
}
err = auth_handler.ProjectInstance(ctx)
if err != nil {
logging.WithFields("instance", instance).OnError(err).Info("trigger auth handler failed")
@@ -299,7 +314,7 @@ func execProjections(ctx context.Context, instances <-chan string, failedInstanc
wg.Done()
}
// returns the instance configured by flag
// queryInstanceIDs returns the instance configured by flag
// or all instances which are not removed
func queryInstanceIDs(ctx context.Context, source *database.DB) []string {
if len(instanceIDs) > 0 {