mirror of
https://github.com/andwati/zitadel.git
synced 2026-10-18 05:52:51 +00:00
fix(eventstore): Make Eventstore Compatible with Relational Table Package (#10687)
Improves compatibility of eventstore and related database components with the new relational table package. ## Which problems are solved 1. **Incompatible Database Interfaces**: The existing eventstore was tightly coupled to the database package, which is incompatible with the new, more abstract relational table package in v3. This prevented the new command-side logic from pushing events to the legacy eventstore. 2. **Missing Health Checks**: The database interfaces in the new package lacked a Ping method, making it impossible to perform health checks on database connections. 3. **Event Publishing Logic**: The command handling logic in domain needed a way to collect and push events to the legacy eventstore after a command was successfully executed. ## How the problems are solved 1. **`LegacyEventstore` Interface**: * A new `LegacyEventstore` interface is introduced in the new `database/eventstore` . This interface exposes a `PushWithNewClient` method that accepts the new `database.QueryExecutor` interface, decoupling the v3 domain from the legacy implementation. * The `internal/eventstore.Eventstore` now implements this interface. A wrapper, PushWithClient, is added to convert the old database client types (`*sql.DB`, `*sql.Tx`) into the new `QueryExecutor` types before calling `PushWithNewClient`. 2. **Database Interface Updates**: * The `database.Pool` and `database.Client` interfaces in `storage/eventstore` have been updated to include a Ping method, allowing for consistent health checks across different database dialects. * The `postgres` and `sql` dialect implementations have been updated to support this new method. 3. **Command and Invoker Refactoring**: * The `Commander` interface in domain now includes an `Events() []legacy_es.Command` method. This allows commands to declare which events they will generate. * The `eventCollector` in the invoker logic has been redesigned. It now ensures a database transaction is started before executing a command. After successful execution, it calls the `Events()` method on the command to collect the generated events and appends them to a list. * The `eventStoreInvoker` then pushes all collected events to the legacy eventstore using the new `LegacyEventstore` interface, ensuring that events are only pushed if the entire command (and any sub-commands) executes successfully within the transaction. 4. **Testing**: * New unit tests have been added for the invoker to verify that events are correctly collected from single commands, batched commands, and nested commands. These changes create a clean bridge between the new v3 command-side logic and the existing v1 eventstore, allowing for incremental adoption of the new architecture while maintaining full functionality. ## Additional Information closes https://github.com/zitadel/zitadel/issues/10442
This commit is contained in:
@@ -2,22 +2,22 @@ package eventstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
_ "embed"
|
||||
"fmt"
|
||||
|
||||
"github.com/riverqueue/river"
|
||||
"github.com/zitadel/logging"
|
||||
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database"
|
||||
"github.com/zitadel/zitadel/backend/v3/storage/database/dialect/sql"
|
||||
"github.com/zitadel/zitadel/internal/api/authz"
|
||||
"github.com/zitadel/zitadel/internal/database"
|
||||
"github.com/zitadel/zitadel/internal/eventstore"
|
||||
"github.com/zitadel/zitadel/internal/queue"
|
||||
exec_repo "github.com/zitadel/zitadel/internal/repository/execution"
|
||||
"github.com/zitadel/zitadel/internal/telemetry/tracing"
|
||||
)
|
||||
|
||||
func (es *Eventstore) Push(ctx context.Context, client database.ContextQueryExecuter, commands ...eventstore.Command) (events []eventstore.Event, err error) {
|
||||
func (es *Eventstore) Push(ctx context.Context, client database.QueryExecutor, commands ...eventstore.Command) (events []eventstore.Event, err error) {
|
||||
ctx, span := tracing.NewSpan(ctx)
|
||||
defer func() { span.EndWithError(err) }()
|
||||
|
||||
@@ -29,20 +29,9 @@ func (es *Eventstore) Push(ctx context.Context, client database.ContextQueryExec
|
||||
return events, err
|
||||
}
|
||||
|
||||
func (es *Eventstore) writeCommands(ctx context.Context, client database.ContextQueryExecuter, commands []eventstore.Command) (_ []eventstore.Event, err error) {
|
||||
var conn *sql.Conn
|
||||
switch c := client.(type) {
|
||||
case database.Client:
|
||||
conn, err = c.Conn(ctx)
|
||||
case nil:
|
||||
conn, err = es.client.Conn(ctx)
|
||||
client = conn
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if conn != nil {
|
||||
defer conn.Close()
|
||||
func (es *Eventstore) writeCommands(ctx context.Context, client database.QueryExecutor, commands []eventstore.Command) (_ []eventstore.Event, err error) {
|
||||
if client == nil {
|
||||
client = es.client
|
||||
}
|
||||
|
||||
tx, close, err := es.pushTx(ctx, client)
|
||||
@@ -56,7 +45,7 @@ func (es *Eventstore) writeCommands(ctx context.Context, client database.Context
|
||||
}
|
||||
|
||||
// lock the instance for reading events if await events is set for the duration of the transaction.
|
||||
_, err = tx.ExecContext(ctx, "SELECT pg_advisory_xact_lock_shared('eventstore.events2'::REGCLASS::OID::INTEGER, hashtext($1))", authz.GetInstance(ctx).InstanceID())
|
||||
_, err = tx.Exec(ctx, "SELECT pg_advisory_xact_lock_shared('eventstore.events2'::REGCLASS::OID::INTEGER, hashtext($1))", authz.GetInstance(ctx).InstanceID())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -82,7 +71,7 @@ func (es *Eventstore) writeCommands(ctx context.Context, client database.Context
|
||||
return events, nil
|
||||
}
|
||||
|
||||
func writeEvents(ctx context.Context, tx database.Tx, commands []eventstore.Command) (_ []eventstore.Event, err error) {
|
||||
func writeEvents(ctx context.Context, tx database.Transaction, commands []eventstore.Command) (_ []eventstore.Event, err error) {
|
||||
ctx, span := tracing.NewSpan(ctx)
|
||||
defer func() { span.EndWithError(err) }()
|
||||
|
||||
@@ -91,7 +80,7 @@ func writeEvents(ctx context.Context, tx database.Tx, commands []eventstore.Comm
|
||||
return nil, err
|
||||
}
|
||||
|
||||
rows, err := tx.QueryContext(ctx, `select owner, created_at, "sequence", position from eventstore.push($1::eventstore.command[])`, cmds)
|
||||
rows, err := tx.Query(ctx, `select owner, created_at, "sequence", position from eventstore.push($1::eventstore.command[])`, cmds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -110,12 +99,12 @@ func writeEvents(ctx context.Context, tx database.Tx, commands []eventstore.Comm
|
||||
return events, nil
|
||||
}
|
||||
|
||||
func (es *Eventstore) queueExecutions(ctx context.Context, tx database.Tx, events []eventstore.Event) error {
|
||||
func (es *Eventstore) queueExecutions(ctx context.Context, tx database.Transaction, events []eventstore.Event) error {
|
||||
if es.queue == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
sqlTx, ok := tx.(*sql.Tx)
|
||||
sqlTx, ok := tx.(*sql.Transaction)
|
||||
if !ok {
|
||||
types := make([]string, len(events))
|
||||
for i, event := range events {
|
||||
@@ -132,7 +121,7 @@ func (es *Eventstore) queueExecutions(ctx context.Context, tx database.Tx, event
|
||||
return nil
|
||||
}
|
||||
return es.queue.InsertManyFastTx(
|
||||
ctx, sqlTx, jobArgs,
|
||||
ctx, sqlTx.Tx, jobArgs,
|
||||
queue.WithQueueName(exec_repo.QueueName),
|
||||
)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user