From cd3ffbd3eb48afeefe50e6f5d1d20445d8cd09d3 Mon Sep 17 00:00:00 2001
From: Silvan <silvan.reusser@gmail.com>
Date: Mon, 12 Aug 2024 12:33:45 +0200
Subject: [PATCH] fix(mirror): use correct statements on push (#8414)

# Which Problems Are Solved

The mirror command used the wrong position to filter for events if
different database technologies for source and destination were used.

# How the Problems Are Solved

The statements which diverge are stored on the client so that different
technologies can use different statements.

# Additional Context

- https://discord.com/channels/927474939156643850/1256396896243552347
---
 cmd/mirror/event_store.go                    |  2 +-
 internal/v2/eventstore/postgres/push.go      |  6 +++---
 internal/v2/eventstore/postgres/push_test.go |  5 ++++-
 internal/v2/eventstore/postgres/storage.go   | 19 ++++++++++---------
 4 files changed, 18 insertions(+), 14 deletions(-)

diff --git a/cmd/mirror/event_store.go b/cmd/mirror/event_store.go
index 358f878d7..23145bdc3 100644
--- a/cmd/mirror/event_store.go
+++ b/cmd/mirror/event_store.go
@@ -44,7 +44,7 @@ Migrate only copies events2 and unique constraints`,
 }
 
 func copyEventstore(ctx context.Context, config *Migration) {
-	sourceClient, err := db.Connect(config.Source, false, dialect.DBPurposeQuery)
+	sourceClient, err := db.Connect(config.Source, false, dialect.DBPurposeEventPusher)
 	logging.OnError(err).Fatal("unable to connect to source database")
 	defer sourceClient.Close()
 
diff --git a/internal/v2/eventstore/postgres/push.go b/internal/v2/eventstore/postgres/push.go
index 0f4c29316..09f663a08 100644
--- a/internal/v2/eventstore/postgres/push.go
+++ b/internal/v2/eventstore/postgres/push.go
@@ -71,7 +71,7 @@ func (s *Storage) Push(ctx context.Context, intent *eventstore.PushIntent) (err
 			return err
 		}
 
-		return push(ctx, tx, intent, commands)
+		return s.push(ctx, tx, intent, commands)
 	})
 }
 
@@ -144,7 +144,7 @@ func lockAggregates(ctx context.Context, tx *sql.Tx, intent *eventstore.PushInte
 	return res, nil
 }
 
-func push(ctx context.Context, tx *sql.Tx, reducer eventstore.Reducer, commands []*command) (err error) {
+func (s *Storage) push(ctx context.Context, tx *sql.Tx, reducer eventstore.Reducer, commands []*command) (err error) {
 	ctx, span := tracing.NewSpan(ctx)
 	defer func() { span.EndWithError(err) }()
 
@@ -171,7 +171,7 @@ func push(ctx context.Context, tx *sql.Tx, reducer eventstore.Reducer, commands
 			cmd.position.InPositionOrder,
 		)
 
-		stmt.WriteString(pushPositionStmt)
+		stmt.WriteString(s.pushPositionStmt)
 		stmt.WriteString(`)`)
 	}
 	stmt.WriteString(` RETURNING created_at, "position"`)
diff --git a/internal/v2/eventstore/postgres/push_test.go b/internal/v2/eventstore/postgres/push_test.go
index 641b36680..91fdc1fcd 100644
--- a/internal/v2/eventstore/postgres/push_test.go
+++ b/internal/v2/eventstore/postgres/push_test.go
@@ -1297,7 +1297,10 @@ func Test_push(t *testing.T) {
 				t.Errorf("unexpected error in begin: %v", err)
 				t.FailNow()
 			}
-			err = push(context.Background(), tx, tt.args.reducer, tt.args.commands)
+			s := Storage{
+				pushPositionStmt: initPushStmt("postgres"),
+			}
+			err = s.push(context.Background(), tx, tt.args.reducer, tt.args.commands)
 			tt.want.assertErr(t, err)
 			dbMock.Assert(t)
 			if tt.args.reducer != nil {
diff --git a/internal/v2/eventstore/postgres/storage.go b/internal/v2/eventstore/postgres/storage.go
index c983cf83f..3a703a7d1 100644
--- a/internal/v2/eventstore/postgres/storage.go
+++ b/internal/v2/eventstore/postgres/storage.go
@@ -12,13 +12,12 @@ import (
 var (
 	_ eventstore.Pusher  = (*Storage)(nil)
 	_ eventstore.Querier = (*Storage)(nil)
-
-	pushPositionStmt string
 )
 
 type Storage struct {
-	client *database.DB
-	config *Config
+	client           *database.DB
+	config           *Config
+	pushPositionStmt string
 }
 
 type Config struct {
@@ -28,19 +27,21 @@ type Config struct {
 func New(client *database.DB, config *Config) *Storage {
 	initPushStmt(client.Type())
 	return &Storage{
-		client: client,
-		config: config,
+		client:           client,
+		config:           config,
+		pushPositionStmt: initPushStmt(client.Type()),
 	}
 }
 
-func initPushStmt(typ string) {
+func initPushStmt(typ string) string {
 	switch typ {
 	case "cockroach":
-		pushPositionStmt = ", hlc_to_timestamp(cluster_logical_timestamp()), cluster_logical_timestamp()"
+		return ", hlc_to_timestamp(cluster_logical_timestamp()), cluster_logical_timestamp()"
 	case "postgres":
-		pushPositionStmt = ", statement_timestamp(), EXTRACT(EPOCH FROM clock_timestamp())"
+		return ", statement_timestamp(), EXTRACT(EPOCH FROM clock_timestamp())"
 	default:
 		logging.WithFields("database_type", typ).Panic("position statement for type not implemented")
+		return ""
 	}
 }