mirror of
https://github.com/andwati/zitadel.git
synced 2026-10-18 05:52:51 +00:00
SilvanandGitHub
22ef817d5c
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
134 lines
3.1 KiB
Go
134 lines
3.1 KiB
Go
package domain
|
|
|
|
import (
|
|
"context"
|
|
|
|
legacy_es "github.com/zitadel/zitadel/internal/eventstore"
|
|
"github.com/zitadel/zitadel/internal/repository/org"
|
|
)
|
|
|
|
// AddOrgCommand adds a new organization.
|
|
// I'm unsure if we should add the Admins here or if this should be a separate command.
|
|
type AddOrgCommand struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
// Admins []*AddMemberCommand `json:"admins"`
|
|
}
|
|
|
|
// func NewAddOrgCommand(name string, admins ...*AddMemberCommand) *AddOrgCommand {
|
|
// return &AddOrgCommand{
|
|
// Name: name,
|
|
// Admins: admins,
|
|
// }
|
|
// }
|
|
|
|
// // String implements [Commander].
|
|
// func (cmd *AddOrgCommand) String() string {
|
|
// return "AddOrgCommand"
|
|
// }
|
|
|
|
// // Execute implements Commander.
|
|
// func (cmd *AddOrgCommand) Execute(ctx context.Context, opts *CommandOpts) (err error) {
|
|
// if len(cmd.Admins) == 0 {
|
|
// return ErrNoAdminSpecified
|
|
// }
|
|
// if err = cmd.ensureID(); err != nil {
|
|
// return err
|
|
// }
|
|
|
|
// close, err := opts.EnsureTx(ctx)
|
|
// if err != nil {
|
|
// return err
|
|
// }
|
|
// defer func() { err = close(ctx, err) }()
|
|
// err = orgRepo(opts.DB).Create(ctx, &Org{
|
|
// ID: cmd.ID,
|
|
// Name: cmd.Name,
|
|
// })
|
|
// if err != nil {
|
|
// return err
|
|
// }
|
|
|
|
// for _, admin := range cmd.Admins {
|
|
// admin.orgID = cmd.ID
|
|
// if err = opts.Invoke(ctx, admin); err != nil {
|
|
// return err
|
|
// }
|
|
// }
|
|
|
|
// orgCache.Set(ctx, &Org{
|
|
// ID: cmd.ID,
|
|
// Name: cmd.Name,
|
|
// })
|
|
|
|
// return nil
|
|
// }
|
|
|
|
// Events implements [eventer].
|
|
func (cmd *AddOrgCommand) Events(ctx context.Context) []legacy_es.Command {
|
|
return []legacy_es.Command{
|
|
org.NewOrgAddedEvent(ctx, &org.NewAggregate(cmd.ID).Aggregate, cmd.Name),
|
|
}
|
|
}
|
|
|
|
// var (
|
|
// _ Commander = (*AddOrgCommand)(nil)
|
|
// _ eventer = (*AddOrgCommand)(nil)
|
|
// )
|
|
|
|
// func (cmd *AddOrgCommand) ensureID() (err error) {
|
|
// if cmd.ID != "" {
|
|
// return nil
|
|
// }
|
|
// cmd.ID, err = generateID()
|
|
// return err
|
|
// }
|
|
|
|
// // AddMemberCommand adds a new member to an organization.
|
|
// // I'm not sure if we should make it more generic to also use it for instances.
|
|
// type AddMemberCommand struct {
|
|
// orgID string
|
|
// UserID string `json:"userId"`
|
|
// Roles []string `json:"roles"`
|
|
// }
|
|
|
|
// func NewAddMemberCommand(userID string, roles ...string) *AddMemberCommand {
|
|
// return &AddMemberCommand{
|
|
// UserID: userID,
|
|
// Roles: roles,
|
|
// }
|
|
// }
|
|
|
|
// // String implements [Commander].
|
|
// func (cmd *AddMemberCommand) String() string {
|
|
// return "AddMemberCommand"
|
|
// }
|
|
|
|
// // Execute implements Commander.
|
|
// func (a *AddMemberCommand) Execute(ctx context.Context, opts *CommandOpts) (err error) {
|
|
// close, err := opts.EnsureTx(ctx)
|
|
// if err != nil {
|
|
// return err
|
|
// }
|
|
// defer func() { err = close(ctx, err) }()
|
|
|
|
// return orgRepo(opts.DB).Member().AddMember(ctx, a.orgID, a.UserID, a.Roles)
|
|
// }
|
|
|
|
// // Events implements [eventer].
|
|
// func (a *AddMemberCommand) Events() []*eventstore.Event {
|
|
// return []*eventstore.Event{
|
|
// {
|
|
// AggregateType: "org",
|
|
// AggregateID: a.UserID,
|
|
// Type: "member.added",
|
|
// Payload: a,
|
|
// },
|
|
// }
|
|
// }
|
|
|
|
// var (
|
|
// _ Commander = (*AddMemberCommand)(nil)
|
|
// _ eventer = (*AddMemberCommand)(nil)
|
|
// )
|