feat(cmd): mirror (#7004)

# Which Problems Are Solved

Adds the possibility to mirror an existing database to a new one. 

For that a new command was added `zitadel mirror`. Including it's
subcommands for a more fine grained mirror of the data.

Sub commands:

* `zitadel mirror eventstore`: copies only events and their unique
constraints
* `zitadel mirror system`: mirrors the data of the `system`-schema
*  `zitadel mirror projections`: runs all projections
*  `zitadel mirror auth`: copies auth requests
* `zitadel mirror verify`: counts the amount of rows in the source and
destination database and prints the diff.

The command requires one of the following flags:
* `--system`: copies all instances of the system
* `--instance <instance-id>`, `--instance <comma separated list of
instance ids>`: copies only the defined instances

The command is save to execute multiple times by adding the
`--replace`-flag. This replaces currently existing data except of the
`events`-table

# Additional Changes

A `--for-mirror`-flag was added to `zitadel setup` to prepare the new
database. The flag skips the creation of the first instances and initial
run of projections.

It is now possible to skip the creation of the first instance during
setup by setting `FirstInstance.Skip` to true in the steps
configuration.

# Additional info

It is currently not possible to merge multiple databases. See
https://github.com/zitadel/zitadel/issues/7964 for more details.

It is currently not possible to use files. See
https://github.com/zitadel/zitadel/issues/7966 for more information.

closes https://github.com/zitadel/zitadel/issues/7586
closes https://github.com/zitadel/zitadel/issues/7486

### Definition of Ready

- [x] I am happy with the code
- [x] Short description of the feature/issue is added in the pr
description
- [x] PR is linked to the corresponding user story
- [x] Acceptance criteria are met
- [x] All open todos and follow ups are defined in a new ticket and
justified
- [x] Deviations from the acceptance criteria and design are agreed with
the PO and documented.
- [x] No debug or dead code
- [x] My code has no repetitions
- [x] Critical parts are tested automatically
- [ ] Where possible E2E tests are implemented
- [x] Documentation/examples are up-to-date
- [x] All non-functional requirements are met
- [x] Functionality of the acceptance criteria is checked manually on
the dev system.

---------

Co-authored-by: Livio Spring <livio.a@gmail.com>

internal/v2/eventstore/postgres/event.go

@@ -29,14 +29,14 @@ func intentToCommands(intent *intent) (commands []*command, err error) {
 }
 
 func marshalPayload(payload any) ([]byte, error) {
-	if reflect.ValueOf(payload).IsZero() {
+	if payload == nil || reflect.ValueOf(payload).IsZero() {
 		return nil, nil
 	}
 	return json.Marshal(payload)
 }
 
 type command struct {
-	eventstore.Command
+	*eventstore.Command
 
 	intent *intent
 

internal/v2/eventstore/postgres/push.go

@@ -3,7 +3,9 @@ package postgres
 import (
 	"context"
 	"database/sql"
+	"fmt"
 
+	"github.com/cockroachdb/cockroach-go/v2/crdb"
 	"github.com/zitadel/logging"
 
 	"github.com/zitadel/zitadel/internal/telemetry/tracing"
@@ -28,40 +30,54 @@ func (s *Storage) Push(ctx context.Context, intent *eventstore.PushIntent) (err
 		}()
 	}
 
-	// allows smaller wait times on query side for instances which are not actively writing
-	if err := setAppName(ctx, tx, "es_pusher_"+intent.Instance()); err != nil {
-		return err
-	}
+	var retryCount uint32
+	return crdb.Execute(func() (err error) {
+		defer func() {
+			if err == nil {
+				return
+			}
+			if retryCount < s.config.MaxRetries {
+				retryCount++
+				return
+			}
+			logging.WithFields("retry_count", retryCount).WithError(err).Debug("max retry count reached")
+			err = zerrors.ThrowInternal(err, "POSTG-VJfJz", "Errors.Internal")
+		}()
+		// allows smaller wait times on query side for instances which are not actively writing
+		if err := setAppName(ctx, tx, "es_pusher_"+intent.Instance()); err != nil {
+			return err
+		}
 
-	intents, err := lockAggregates(ctx, tx, intent)
-	if err != nil {
-		return err
-	}
-
-	if !checkSequences(intents) {
-		return zerrors.ThrowInvalidArgument(nil, "POSTG-KOM6E", "Errors.Internal.Eventstore.SequenceNotMatched")
-	}
-
-	commands := make([]*command, 0, len(intents))
-	for _, intent := range intents {
-		additionalCommands, err := intentToCommands(intent)
+		intents, err := lockAggregates(ctx, tx, intent)
 		if err != nil {
 			return err
 		}
-		commands = append(commands, additionalCommands...)
-	}
 
-	err = uniqueConstraints(ctx, tx, commands)
-	if err != nil {
-		return err
-	}
+		if !checkSequences(intents) {
+			return zerrors.ThrowInvalidArgument(nil, "POSTG-KOM6E", "Errors.Internal.Eventstore.SequenceNotMatched")
+		}
 
-	return push(ctx, tx, intent, commands)
+		commands := make([]*command, 0, len(intents))
+		for _, intent := range intents {
+			additionalCommands, err := intentToCommands(intent)
+			if err != nil {
+				return err
+			}
+			commands = append(commands, additionalCommands...)
+		}
+
+		err = uniqueConstraints(ctx, tx, commands)
+		if err != nil {
+			return err
+		}
+
+		return push(ctx, tx, intent, commands)
+	})
 }
 
 // setAppName for the the current transaction
 func setAppName(ctx context.Context, tx *sql.Tx, name string) error {
-	_, err := tx.ExecContext(ctx, "SET LOCAL application_name TO $1", name)
+	_, err := tx.ExecContext(ctx, fmt.Sprintf("SET LOCAL application_name TO '%s'", name))
 	if err != nil {
 		logging.WithFields("name", name).WithError(err).Debug("setting app name failed")
 		return zerrors.ThrowInternal(err, "POSTG-G3OmZ", "Errors.Internal")
@@ -154,7 +170,8 @@ func push(ctx context.Context, tx *sql.Tx, reducer eventstore.Reducer, commands
 			cmd.sequence,
 			cmd.position.InPositionOrder,
 		)
-		stmt.WriteString(", statement_timestamp(), EXTRACT(EPOCH FROM clock_timestamp())")
+
+		stmt.WriteString(pushPositionStmt)
 		stmt.WriteString(`)`)
 	}
 	stmt.WriteString(` RETURNING created_at, "position"`)

internal/v2/eventstore/postgres/push_test.go

@@ -36,7 +36,9 @@ func Test_uniqueConstraints(t *testing.T) {
 			name: "command without constraints",
 			args: args{
 				commands: []*command{
-					{},
+					{
+						Command: &eventstore.Command{},
+					},
 				},
 				expectations: []mock.Expectation{},
 			},
@@ -53,7 +55,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewAddEventUniqueConstraint("test", "id", "error"),
 							},
@@ -81,7 +83,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewAddGlobalUniqueConstraint("test", "id", "error"),
 							},
@@ -109,7 +111,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewAddEventUniqueConstraint("test", "id", "error"),
 								eventstore.NewAddEventUniqueConstraint("test", "id2", "error"),
@@ -143,7 +145,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewAddEventUniqueConstraint("test", "id", "error"),
 							},
@@ -156,7 +158,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewAddEventUniqueConstraint("test", "id2", "error"),
 							},
@@ -189,7 +191,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewRemoveInstanceUniqueConstraints(),
 							},
@@ -217,7 +219,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewRemoveInstanceUniqueConstraints(),
 							},
@@ -230,7 +232,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewRemoveInstanceUniqueConstraints(),
 							},
@@ -263,7 +265,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewRemoveUniqueConstraint("test", "id"),
 							},
@@ -291,7 +293,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewRemoveGlobalUniqueConstraint("test", "id"),
 							},
@@ -319,7 +321,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewRemoveUniqueConstraint("test", "id"),
 								eventstore.NewRemoveUniqueConstraint("test", "id2"),
@@ -353,7 +355,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewRemoveUniqueConstraint("test", "id"),
 							},
@@ -366,7 +368,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewRemoveUniqueConstraint("test", "id2"),
 							},
@@ -399,7 +401,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewAddEventUniqueConstraint("test", "id", ""),
 							},
@@ -433,7 +435,7 @@ func Test_uniqueConstraints(t *testing.T) {
 								eventstore.AppendAggregate("", "", ""),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							UniqueConstraints: []*eventstore.UniqueConstraint{
 								eventstore.NewAddEventUniqueConstraint("test", "id", "My.Error"),
 							},
@@ -786,7 +788,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -841,7 +843,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -857,7 +859,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -926,7 +928,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -942,7 +944,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "type2", "id2"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -1011,7 +1013,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -1067,7 +1069,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -1123,7 +1125,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -1139,7 +1141,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -1214,7 +1216,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -1230,7 +1232,7 @@ func Test_push(t *testing.T) {
 								eventstore.AppendAggregate("owner", "testType", "testID"),
 							).Aggregates()[0],
 						},
-						Command: eventstore.Command{
+						Command: &eventstore.Command{
 							Action: eventstore.Action[any]{
 								Creator:  "gigi",
 								Revision: 1,
@@ -1286,6 +1288,7 @@ func Test_push(t *testing.T) {
 			},
 		},
 	}
+	initPushStmt("postgres")
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			dbMock := mock.NewSQLMock(t, append([]mock.Expectation{mock.ExpectBegin(nil)}, tt.args.expectations...)...)

internal/v2/eventstore/postgres/query.go

@@ -194,6 +194,7 @@ func writeAggregateFilters(stmt *database.Statement, filters []*eventstore.Aggre
 
 func writeAggregateFilter(stmt *database.Statement, filter *eventstore.AggregateFilter) {
 	conditions := definedConditions([]*condition{
+		{column: "owner", condition: filter.Owners()},
 		{column: "aggregate_type", condition: filter.Type()},
 		{column: "aggregate_id", condition: filter.IDs()},
 	})

internal/v2/eventstore/postgres/storage.go

@@ -3,6 +3,8 @@ package postgres
 import (
 	"context"
 
+	"github.com/zitadel/logging"
+
 	"github.com/zitadel/zitadel/internal/database"
 	"github.com/zitadel/zitadel/internal/v2/eventstore"
 )
@@ -10,15 +12,35 @@ import (
 var (
 	_ eventstore.Pusher  = (*Storage)(nil)
 	_ eventstore.Querier = (*Storage)(nil)
+
+	pushPositionStmt string
 )
 
 type Storage struct {
 	client *database.DB
+	config *Config
 }
 
-func New(client *database.DB) *Storage {
+type Config struct {
+	MaxRetries uint32
+}
+
+func New(client *database.DB, config *Config) *Storage {
+	initPushStmt(client.Type())
 	return &Storage{
 		client: client,
+		config: config,
+	}
+}
+
+func initPushStmt(typ string) {
+	switch typ {
+	case "cockroach":
+		pushPositionStmt = ", hlc_to_timestamp(cluster_logical_timestamp()), cluster_logical_timestamp()"
+	case "postgres":
+		pushPositionStmt = ", statement_timestamp(), EXTRACT(EPOCH FROM clock_timestamp())"
+	default:
+		logging.WithFields("database_type", typ).Panic("position statement for type not implemented")
 	}
 }
 

internal/v2/eventstore/push.go

@@ -87,7 +87,7 @@ type PushAggregate struct {
 	// owner of the aggregate
 	owner string
 	// Commands is an ordered list of changes on the aggregate
-	commands []Command
+	commands []*Command
 	// CurrentSequence checks the current state of the aggregate.
 	// The following types match the current sequence of the aggregate as described:
 	// * nil or [SequenceIgnore]: Not relevant to add the commands
@@ -122,7 +122,7 @@ func (pa *PushAggregate) Owner() string {
 	return pa.owner
 }
 
-func (pa *PushAggregate) Commands() []Command {
+func (pa *PushAggregate) Commands() []*Command {
 	return pa.commands
 }
 
@@ -165,7 +165,7 @@ func CurrentSequenceAtLeast(sequence uint32) PushAggregateOpt {
 	}
 }
 
-func AppendCommands(commands ...Command) PushAggregateOpt {
+func AppendCommands(commands ...*Command) PushAggregateOpt {
 	return func(pa *PushAggregate) {
 		pa.commands = append(pa.commands, commands...)
 	}

internal/v2/eventstore/query.go

@@ -255,6 +255,7 @@ func NewAggregateFilter(typ string, opts ...AggregateFilterOpt) *AggregateFilter
 type AggregateFilter struct {
 	typ    string
 	ids    []string
+	owners *filter[[]string]
 	events []*EventFilter
 }
 
@@ -273,6 +274,13 @@ func (f *AggregateFilter) IDs() database.Condition {
 	return database.NewListContains(f.ids...)
 }
 
+func (f *AggregateFilter) Owners() database.Condition {
+	if f.owners == nil {
+		return nil
+	}
+	return f.owners.condition
+}
+
 func (f *AggregateFilter) Events() []*EventFilter {
 	return f.events
 }
@@ -298,6 +306,61 @@ func AggregateIDs(ids ...string) AggregateFilterOpt {
 	}
 }
 
+func AggregateOwnersEqual(owners ...string) AggregateFilterOpt {
+	return func(f *AggregateFilter) {
+		var cond database.Condition
+		switch len(owners) {
+		case 0:
+			return
+		case 1:
+			cond = database.NewTextEqual(owners[0])
+		default:
+			cond = database.NewListEquals(owners...)
+		}
+		f.owners = &filter[[]string]{
+			condition: cond,
+			value:     &owners,
+		}
+	}
+}
+
+func AggregateOwnersContains(owners ...string) AggregateFilterOpt {
+	return func(f *AggregateFilter) {
+		var cond database.Condition
+		switch len(owners) {
+		case 0:
+			return
+		case 1:
+			cond = database.NewTextEqual(owners[0])
+		default:
+			cond = database.NewListContains(owners...)
+		}
+
+		f.owners = &filter[[]string]{
+			condition: cond,
+			value:     &owners,
+		}
+	}
+}
+
+func AggregateOwnersNotContains(owners ...string) AggregateFilterOpt {
+	return func(f *AggregateFilter) {
+		var cond database.Condition
+		switch len(owners) {
+		case 0:
+			return
+		case 1:
+			cond = database.NewTextUnequal(owners[0])
+		default:
+			cond = database.NewListNotContains(owners...)
+		}
+		f.owners = &filter[[]string]{
+			condition: cond,
+			value:     &owners,
+		}
+	}
+}
+
 func AppendEvent(opts ...EventFilterOpt) AggregateFilterOpt {
 	return AppendEvents(NewEventFilter(opts...))
 }

internal/v2/projection/highest_position.go

@@ -0,0 +1,15 @@
+package projection
+
+import (
+	"github.com/zitadel/zitadel/internal/v2/eventstore"
+)
+
+type HighestPosition eventstore.GlobalPosition
+
+var _ eventstore.Reducer = (*HighestPosition)(nil)
+
+// Reduce implements eventstore.Reducer.
+func (h *HighestPosition) Reduce(events ...*eventstore.StorageEvent) error {
+	*h = HighestPosition(events[len(events)-1].Position)
+	return nil
+}

internal/v2/readmodel/last_successful_mirror.go

@@ -0,0 +1,72 @@
+package readmodel
+
+import (
+	"github.com/zitadel/zitadel/internal/v2/eventstore"
+	"github.com/zitadel/zitadel/internal/v2/system"
+	"github.com/zitadel/zitadel/internal/v2/system/mirror"
+)
+
+type LastSuccessfulMirror struct {
+	ID       string
+	Position float64
+	source   string
+}
+
+func NewLastSuccessfulMirror(source string) *LastSuccessfulMirror {
+	return &LastSuccessfulMirror{
+		source: source,
+	}
+}
+
+var _ eventstore.Reducer = (*LastSuccessfulMirror)(nil)
+
+func (p *LastSuccessfulMirror) Filter() *eventstore.Filter {
+	return eventstore.NewFilter(
+		eventstore.AppendAggregateFilter(
+			system.AggregateType,
+			eventstore.AggregateOwnersEqual(system.AggregateOwner),
+			eventstore.AppendEvent(
+				eventstore.SetEventTypes(
+					mirror.SucceededType,
+				),
+				eventstore.EventCreatorsEqual(mirror.Creator),
+			),
+		),
+		eventstore.FilterPagination(
+			eventstore.Descending(),
+		),
+	)
+}
+
+// Reduce implements eventstore.Reducer.
+func (h *LastSuccessfulMirror) Reduce(events ...*eventstore.StorageEvent) (err error) {
+	for _, event := range events {
+		if event.Type == mirror.SucceededType {
+			err = h.reduceSucceeded(event)
+		}
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (h *LastSuccessfulMirror) reduceSucceeded(event *eventstore.StorageEvent) error {
+	// if position is set we skip all older events
+	if h.Position > 0 {
+		return nil
+
+	}
+	succeededEvent, err := mirror.SucceededEventFromStorage(event)
+	if err != nil {
+		return err
+	}
+
+	if h.source != succeededEvent.Payload.Source {
+		return nil
+	}
+
+	h.Position = succeededEvent.Payload.Position
+
+	return nil
+}

internal/v2/system/aggregate.go

@@ -0,0 +1,8 @@
+package system
+
+const (
+	AggregateType     = "system"
+	AggregateOwner    = "SYSTEM"
+	AggregateInstance = ""
+	EventTypePrefix   = AggregateType + "."
+)

internal/v2/system/mirror/aggregate.go

@@ -0,0 +1,8 @@
+package mirror
+
+import "github.com/zitadel/zitadel/internal/v2/system"
+
+const (
+	Creator         = "MIRROR"
+	eventTypePrefix = system.EventTypePrefix + "mirror."
+)

internal/v2/system/mirror/failed.go

@@ -0,0 +1,52 @@
+package mirror
+
+import (
+	"github.com/zitadel/zitadel/internal/v2/eventstore"
+	"github.com/zitadel/zitadel/internal/zerrors"
+)
+
+type failedPayload struct {
+	Cause string `json:"cause"`
+	// Source is the name of the database data are mirrored to
+	Source string `json:"source"`
+}
+
+const FailedType = eventTypePrefix + "failed"
+
+type FailedEvent eventstore.Event[failedPayload]
+
+var _ eventstore.TypeChecker = (*FailedEvent)(nil)
+
+func (e *FailedEvent) ActionType() string {
+	return FailedType
+}
+
+func FailedEventFromStorage(event *eventstore.StorageEvent) (e *FailedEvent, _ error) {
+	if event.Type != e.ActionType() {
+		return nil, zerrors.ThrowInvalidArgument(nil, "MIRRO-bwB9l", "Errors.Invalid.Event.Type")
+	}
+
+	payload, err := eventstore.UnmarshalPayload[failedPayload](event.Payload)
+	if err != nil {
+		return nil, err
+	}
+
+	return &FailedEvent{
+		StorageEvent: event,
+		Payload:      payload,
+	}, nil
+}
+
+func NewFailedCommand(source string, cause error) *eventstore.Command {
+	return &eventstore.Command{
+		Action: eventstore.Action[any]{
+			Creator: Creator,
+			Type:    FailedType,
+			Payload: failedPayload{
+				Cause:  cause.Error(),
+				Source: source,
+			},
+			Revision: 1,
+		},
+	}
+}

internal/v2/system/mirror/started.go

@@ -0,0 +1,68 @@
+package mirror
+
+import (
+	"github.com/zitadel/zitadel/internal/v2/eventstore"
+	"github.com/zitadel/zitadel/internal/zerrors"
+)
+
+type startedPayload struct {
+	// Destination is the name of the database data are mirrored to
+	Destination string `json:"destination"`
+	// Either Instances or System needs to be set
+	Instances []string `json:"instances,omitempty"`
+	System    bool     `json:"system,omitempty"`
+}
+
+const StartedType = eventTypePrefix + "started"
+
+type StartedEvent eventstore.Event[startedPayload]
+
+var _ eventstore.TypeChecker = (*StartedEvent)(nil)
+
+func (e *StartedEvent) ActionType() string {
+	return StartedType
+}
+
+func StartedEventFromStorage(event *eventstore.StorageEvent) (e *StartedEvent, _ error) {
+	if event.Type != e.ActionType() {
+		return nil, zerrors.ThrowInvalidArgument(nil, "MIRRO-bwB9l", "Errors.Invalid.Event.Type")
+	}
+
+	payload, err := eventstore.UnmarshalPayload[startedPayload](event.Payload)
+	if err != nil {
+		return nil, err
+	}
+
+	return &StartedEvent{
+		StorageEvent: event,
+		Payload:      payload,
+	}, nil
+}
+
+func NewStartedSystemCommand(destination string) *eventstore.Command {
+	return newStartedCommand(&startedPayload{
+		Destination: destination,
+		System:      true,
+	})
+}
+
+func NewStartedInstancesCommand(destination string, instances []string) (*eventstore.Command, error) {
+	if len(instances) == 0 {
+		return nil, zerrors.ThrowInvalidArgument(nil, "MIRRO-8YkrE", "Errors.Mirror.NoInstances")
+	}
+	return newStartedCommand(&startedPayload{
+		Destination: destination,
+		Instances:   instances,
+	}), nil
+}
+
+func newStartedCommand(payload *startedPayload) *eventstore.Command {
+	return &eventstore.Command{
+		Action: eventstore.Action[any]{
+			Creator:  Creator,
+			Type:     StartedType,
+			Revision: 1,
+			Payload:  *payload,
+		},
+	}
+}

internal/v2/system/mirror/succeeded.go

@@ -0,0 +1,53 @@
+package mirror
+
+import (
+	"github.com/zitadel/zitadel/internal/v2/eventstore"
+	"github.com/zitadel/zitadel/internal/zerrors"
+)
+
+type succeededPayload struct {
+	// Source is the name of the database data are mirrored from
+	Source string `json:"source"`
+	// Position until data will be mirrored
+	Position float64 `json:"position"`
+}
+
+const SucceededType = eventTypePrefix + "succeeded"
+
+type SucceededEvent eventstore.Event[succeededPayload]
+
+var _ eventstore.TypeChecker = (*SucceededEvent)(nil)
+
+func (e *SucceededEvent) ActionType() string {
+	return SucceededType
+}
+
+func SucceededEventFromStorage(event *eventstore.StorageEvent) (e *SucceededEvent, _ error) {
+	if event.Type != e.ActionType() {
+		return nil, zerrors.ThrowInvalidArgument(nil, "MIRRO-xh5IW", "Errors.Invalid.Event.Type")
+	}
+
+	payload, err := eventstore.UnmarshalPayload[succeededPayload](event.Payload)
+	if err != nil {
+		return nil, err
+	}
+
+	return &SucceededEvent{
+		StorageEvent: event,
+		Payload:      payload,
+	}, nil
+}
+
+func NewSucceededCommand(source string, position float64) *eventstore.Command {
+	return &eventstore.Command{
+		Action: eventstore.Action[any]{
+			Creator:  Creator,
+			Type:     SucceededType,
+			Revision: 1,
+			Payload: succeededPayload{
+				Source:   source,
+				Position: position,
+			},
+		},
+	}
+}