From 3aba942162676c9f9c65d74ffa6f61820e73195e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tim=20M=C3=B6hlmann?= Date: Wed, 11 Sep 2024 11:24:00 +0300 Subject: [PATCH] feat: add debug events API (#8533) # Which Problems Are Solved Add a debug API which allows pushing a set of events to be reduced in a dedicated projection. The events can carry a sleep duration which simulates a slow query during projection handling. # How the Problems Are Solved - `CreateDebugEvents` allows pushing multiple events which simulate the lifecycle of a resource. Each event has a `projectionSleep` field, which issues a `pg_sleep()` statement query in the projection handler : - Add - Change - Remove - `ListDebugEventsStates` list the current state of the projection, optionally with a Trigger - `GetDebugEventsStateByID` get the current state of the aggregate ID in the projection, optionally with a Trigger # Additional Changes - none # Additional Context - Allows reproduction of https://github.com/zitadel/zitadel/issues/8517 --- cmd/defaults.yaml | 3 + cmd/start/start.go | 4 + .../debug_events/debug_events/debug.go | 51 +++ .../debug_events/debug_converter.go | 63 ++++ .../debug_events/debug_events/server.go | 47 +++ internal/command/debug_events.go | 82 +++++ internal/command/debug_events_model.go | 68 ++++ internal/command/debug_events_test.go | 340 ++++++++++++++++++ internal/domain/debug_events.go | 14 + internal/eventstore/handler/v2/statement.go | 21 ++ internal/query/debug_events.go | 106 ++++++ internal/query/debug_events_state_by_id.sql | 5 + internal/query/debug_events_states.sql | 4 + internal/query/projection/debug_events.go | 146 ++++++++ internal/query/projection/projection.go | 3 + internal/repository/debug_events/aggregate.go | 27 ++ internal/repository/debug_events/debug.go | 125 +++++++ .../repository/debug_events/eventstore.go | 11 + .../v3alpha/debug_events_service.proto | 223 ++++++++++++ .../debug_events/v3alpha/event.proto | 47 +++ .../debug_events/v3alpha/state.proto | 14 + 21 files changed, 1404 insertions(+) create mode 100644 internal/api/grpc/resources/debug_events/debug_events/debug.go create mode 100644 internal/api/grpc/resources/debug_events/debug_events/debug_converter.go create mode 100644 internal/api/grpc/resources/debug_events/debug_events/server.go create mode 100644 internal/command/debug_events.go create mode 100644 internal/command/debug_events_model.go create mode 100644 internal/command/debug_events_test.go create mode 100644 internal/domain/debug_events.go create mode 100644 internal/query/debug_events.go create mode 100644 internal/query/debug_events_state_by_id.sql create mode 100644 internal/query/debug_events_states.sql create mode 100644 internal/query/projection/debug_events.go create mode 100644 internal/repository/debug_events/aggregate.go create mode 100644 internal/repository/debug_events/debug.go create mode 100644 internal/repository/debug_events/eventstore.go create mode 100644 proto/zitadel/resources/debug_events/v3alpha/debug_events_service.proto create mode 100644 proto/zitadel/resources/debug_events/v3alpha/event.proto create mode 100644 proto/zitadel/resources/debug_events/v3alpha/state.proto diff --git a/cmd/defaults.yaml b/cmd/defaults.yaml index 76f2fb7fb..f36932b02 100644 --- a/cmd/defaults.yaml +++ b/cmd/defaults.yaml @@ -1033,6 +1033,8 @@ InternalAuthZ: - "iam.web_key.write" - "iam.web_key.delete" - "iam.web_key.read" + - "iam.debug.write" + - "iam.debug.read" - "org.read" - "org.global.read" - "org.create" @@ -1110,6 +1112,7 @@ InternalAuthZ: - "iam.restrictions.read" - "iam.feature.read" - "iam.web_key.read" + - "iam.debug.read" - "org.read" - "org.member.read" - "org.idp.read" diff --git a/cmd/start/start.go b/cmd/start/start.go index 594e16776..5fc4ba936 100644 --- a/cmd/start/start.go +++ b/cmd/start/start.go @@ -45,6 +45,7 @@ import ( org_v2 "github.com/zitadel/zitadel/internal/api/grpc/org/v2" org_v2beta "github.com/zitadel/zitadel/internal/api/grpc/org/v2beta" action_v3_alpha "github.com/zitadel/zitadel/internal/api/grpc/resources/action/v3alpha" + "github.com/zitadel/zitadel/internal/api/grpc/resources/debug_events/debug_events" user_v3_alpha "github.com/zitadel/zitadel/internal/api/grpc/resources/user/v3alpha" userschema_v3_alpha "github.com/zitadel/zitadel/internal/api/grpc/resources/userschema/v3alpha" "github.com/zitadel/zitadel/internal/api/grpc/resources/webkey/v3" @@ -459,6 +460,9 @@ func startAPIs( if err := apis.RegisterService(ctx, webkey.CreateServer(commands, queries)); err != nil { return nil, err } + if err := apis.RegisterService(ctx, debug_events.CreateServer(commands, queries)); err != nil { + return nil, err + } instanceInterceptor := middleware.InstanceInterceptor(queries, config.ExternalDomain, login.IgnoreInstanceEndpoints...) assetsCache := middleware.AssetsCacheInterceptor(config.AssetStorage.Cache.MaxAge, config.AssetStorage.Cache.SharedMaxAge) apis.RegisterHandlerOnPrefix(assets.HandlerPrefix, assets.NewHandler(commands, verifier, config.InternalAuthZ, id.SonyFlakeGenerator(), store, queries, middleware.CallDurationHandler, instanceInterceptor.Handler, assetsCache.Handler, limitingAccessInterceptor.Handle)) diff --git a/internal/api/grpc/resources/debug_events/debug_events/debug.go b/internal/api/grpc/resources/debug_events/debug_events/debug.go new file mode 100644 index 000000000..109165ffb --- /dev/null +++ b/internal/api/grpc/resources/debug_events/debug_events/debug.go @@ -0,0 +1,51 @@ +package debug_events + +import ( + "context" + + "github.com/zitadel/zitadel/internal/api/authz" + resource_object "github.com/zitadel/zitadel/internal/api/grpc/resources/object/v3alpha" + "github.com/zitadel/zitadel/internal/telemetry/tracing" + object "github.com/zitadel/zitadel/pkg/grpc/object/v3alpha" + debug_events "github.com/zitadel/zitadel/pkg/grpc/resources/debug_events/v3alpha" +) + +func (s *Server) CreateDebugEvents(ctx context.Context, req *debug_events.CreateDebugEventsRequest) (_ *debug_events.CreateDebugEventsResponse, err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + + details, err := s.command.CreateDebugEvents(ctx, debugEventsFromRequest(req)) + if err != nil { + return nil, err + } + return &debug_events.CreateDebugEventsResponse{ + Details: resource_object.DomainToDetailsPb(details, object.OwnerType_OWNER_TYPE_INSTANCE, authz.GetInstance(ctx).InstanceID()), + }, nil +} + +func (s *Server) GetDebugEventsStateById(ctx context.Context, req *debug_events.GetDebugEventsStateByIdRequest) (_ *debug_events.GetDebugEventsStateByIdResponse, err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + + state, err := s.query.GetDebugEventsStateByID(ctx, req.GetId(), req.GetTriggerBulk()) + if err != nil { + return nil, err + } + + return &debug_events.GetDebugEventsStateByIdResponse{ + State: eventsStateToPB(state), + }, nil +} +func (s *Server) ListDebugEventsStates(ctx context.Context, req *debug_events.ListDebugEventsStatesRequest) (_ *debug_events.ListDebugEventsStatesResponse, err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + + states, err := s.query.ListDebugEventsStates(ctx, req.GetTriggerBulk()) + if err != nil { + return nil, err + } + + return &debug_events.ListDebugEventsStatesResponse{ + States: eventStatesToPB(states), + }, nil +} diff --git a/internal/api/grpc/resources/debug_events/debug_events/debug_converter.go b/internal/api/grpc/resources/debug_events/debug_events/debug_converter.go new file mode 100644 index 000000000..9924d7813 --- /dev/null +++ b/internal/api/grpc/resources/debug_events/debug_events/debug_converter.go @@ -0,0 +1,63 @@ +package debug_events + +import ( + "fmt" + + resource_object "github.com/zitadel/zitadel/internal/api/grpc/resources/object/v3alpha" + "github.com/zitadel/zitadel/internal/command" + "github.com/zitadel/zitadel/internal/query" + object "github.com/zitadel/zitadel/pkg/grpc/object/v3alpha" + debug_events "github.com/zitadel/zitadel/pkg/grpc/resources/debug_events/v3alpha" +) + +func debugEventsFromRequest(req *debug_events.CreateDebugEventsRequest) *command.DebugEvents { + reqEvents := req.GetEvents() + events := make([]command.DebugEvent, len(reqEvents)) + for i, event := range reqEvents { + events[i] = debugEventFromRequest(event) + } + + return &command.DebugEvents{ + AggregateID: req.GetAggregateId(), + Events: events, + } +} + +func debugEventFromRequest(event *debug_events.Event) command.DebugEvent { + switch e := event.Event.(type) { + case *debug_events.Event_Add: + return command.DebugEventAdded{ + ProjectionSleep: e.Add.GetProjectionSleep().AsDuration(), + Blob: e.Add.Blob, + } + + case *debug_events.Event_Change: + return command.DebugEventChanged{ + ProjectionSleep: e.Change.GetProjectionSleep().AsDuration(), + Blob: e.Change.Blob, + } + + case *debug_events.Event_Remove: + return command.DebugEventRemoved{ + ProjectionSleep: e.Remove.GetProjectionSleep().AsDuration(), + } + + default: + panic(fmt.Errorf("invalid debug event type %T", event.Event)) + } +} + +func eventsStateToPB(state *query.DebugEventState) *debug_events.State { + return &debug_events.State{ + Details: resource_object.DomainToDetailsPb(&state.ObjectDetails, object.OwnerType_OWNER_TYPE_INSTANCE, state.ResourceOwner), + Blob: state.Blob, + } +} + +func eventStatesToPB(states []query.DebugEventState) []*debug_events.State { + out := make([]*debug_events.State, len(states)) + for i, state := range states { + out[i] = eventsStateToPB(&state) + } + return out +} diff --git a/internal/api/grpc/resources/debug_events/debug_events/server.go b/internal/api/grpc/resources/debug_events/debug_events/server.go new file mode 100644 index 000000000..82401f214 --- /dev/null +++ b/internal/api/grpc/resources/debug_events/debug_events/server.go @@ -0,0 +1,47 @@ +package debug_events + +import ( + "google.golang.org/grpc" + + "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/api/grpc/server" + "github.com/zitadel/zitadel/internal/command" + "github.com/zitadel/zitadel/internal/query" + debug_events "github.com/zitadel/zitadel/pkg/grpc/resources/debug_events/v3alpha" +) + +type Server struct { + debug_events.UnimplementedZITADELDebugEventsServer + command *command.Commands + query *query.Queries +} + +func CreateServer( + command *command.Commands, + query *query.Queries, +) *Server { + return &Server{ + command: command, + query: query, + } +} + +func (s *Server) RegisterServer(grpcServer *grpc.Server) { + debug_events.RegisterZITADELDebugEventsServer(grpcServer, s) +} + +func (s *Server) AppName() string { + return debug_events.ZITADELDebugEvents_ServiceDesc.ServiceName +} + +func (s *Server) MethodPrefix() string { + return debug_events.ZITADELDebugEvents_ServiceDesc.ServiceName +} + +func (s *Server) AuthMethods() authz.MethodMapping { + return debug_events.ZITADELDebugEvents_AuthMethods +} + +func (s *Server) RegisterGateway() server.RegisterGatewayFunc { + return debug_events.RegisterZITADELDebugEventsHandler +} diff --git a/internal/command/debug_events.go b/internal/command/debug_events.go new file mode 100644 index 000000000..50f87bd2f --- /dev/null +++ b/internal/command/debug_events.go @@ -0,0 +1,82 @@ +package command + +import ( + "context" + "time" + + "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/domain" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/repository/debug_events" + "github.com/zitadel/zitadel/internal/telemetry/tracing" + "github.com/zitadel/zitadel/internal/zerrors" +) + +type DebugEvents struct { + AggregateID string + Events []DebugEvent +} + +type DebugEvent interface { + isADebugEvent() +} + +type DebugEventAdded struct { + ProjectionSleep time.Duration + Blob *string +} + +type DebugEventChanged struct { + ProjectionSleep time.Duration + Blob *string +} + +type DebugEventRemoved struct { + ProjectionSleep time.Duration +} + +func (DebugEventAdded) isADebugEvent() {} +func (DebugEventChanged) isADebugEvent() {} +func (DebugEventRemoved) isADebugEvent() {} + +func (c *Commands) CreateDebugEvents(ctx context.Context, dbe *DebugEvents) (_ *domain.ObjectDetails, err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + + model := NewDebugEventsWriteModel(dbe.AggregateID, authz.GetInstance(ctx).InstanceID()) + if err = c.eventstore.FilterToQueryReducer(ctx, model); err != nil { + return nil, err + } + aggr := debug_events.AggregateFromWriteModel(ctx, &model.WriteModel) + + cmds := make([]eventstore.Command, len(dbe.Events)) + for i, event := range dbe.Events { + var cmd eventstore.Command + switch e := event.(type) { + case DebugEventAdded: + if model.State.Exists() { + return nil, zerrors.ThrowAlreadyExists(nil, "COMMAND-Aex6j", "debug aggregate already exists") + } + cmd = debug_events.NewAddedEvent(ctx, aggr, e.ProjectionSleep, e.Blob) + case DebugEventChanged: + if !model.State.Exists() { + return nil, zerrors.ThrowNotFound(nil, "COMMAND-Thie6", "debug aggregate not found") + } + cmd = debug_events.NewChangedEvent(ctx, aggr, e.ProjectionSleep, e.Blob) + case DebugEventRemoved: + if !model.State.Exists() { + return nil, zerrors.ThrowNotFound(nil, "COMMAND-Ohna9", "debug aggregate not found") + } + cmd = debug_events.NewRemovedEvent(ctx, aggr, e.ProjectionSleep) + } + + cmds[i] = cmd + // be sure the state of the last event is reduced before handling the next one. + model.reduceEvent(cmd.(eventstore.Event)) + } + events, err := c.eventstore.Push(ctx, cmds...) + if err != nil { + return nil, err + } + return pushedEventsToObjectDetails(events), nil +} diff --git a/internal/command/debug_events_model.go b/internal/command/debug_events_model.go new file mode 100644 index 000000000..81d0aca6f --- /dev/null +++ b/internal/command/debug_events_model.go @@ -0,0 +1,68 @@ +package command + +import ( + "github.com/zitadel/zitadel/internal/domain" + "github.com/zitadel/zitadel/internal/eventstore" + debug "github.com/zitadel/zitadel/internal/repository/debug_events" +) + +type DebugEventsWriteModel struct { + eventstore.WriteModel + State domain.DebugEventsState + Blob string +} + +func NewDebugEventsWriteModel(aggregateID, resourceOwner string) *DebugEventsWriteModel { + return &DebugEventsWriteModel{ + WriteModel: eventstore.WriteModel{ + AggregateID: aggregateID, + ResourceOwner: resourceOwner, + }, + } +} + +func (wm *DebugEventsWriteModel) AppendEvents(events ...eventstore.Event) { + wm.WriteModel.AppendEvents(events...) +} + +func (wm *DebugEventsWriteModel) Reduce() error { + for _, event := range wm.Events { + wm.reduceEvent(event) + } + return wm.WriteModel.Reduce() +} + +func (wm *DebugEventsWriteModel) reduceEvent(event eventstore.Event) { + if event.Aggregate().ID != wm.AggregateID { + return + } + switch e := event.(type) { + case *debug.AddedEvent: + wm.State = domain.DebugEventsStateInitial + if e.Blob != nil { + wm.Blob = *e.Blob + } + case *debug.ChangedEvent: + wm.State = domain.DebugEventsStateChanged + if e.Blob != nil { + wm.Blob = *e.Blob + } + case *debug.RemovedEvent: + wm.State = domain.DebugEventsStateRemoved + wm.Blob = "" + } +} + +func (wm *DebugEventsWriteModel) Query() *eventstore.SearchQueryBuilder { + return eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent). + ResourceOwner(wm.ResourceOwner). + AddQuery(). + AggregateTypes(debug.AggregateType). + AggregateIDs(wm.AggregateID). + EventTypes( + debug.AddedEventType, + debug.ChangedEventType, + debug.RemovedEventType, + ). + Builder() +} diff --git a/internal/command/debug_events_test.go b/internal/command/debug_events_test.go new file mode 100644 index 000000000..e740de857 --- /dev/null +++ b/internal/command/debug_events_test.go @@ -0,0 +1,340 @@ +package command + +import ( + "io" + "testing" + "time" + + "github.com/muhlemmer/gu" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/domain" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/repository/debug_events" + "github.com/zitadel/zitadel/internal/zerrors" +) + +func TestCommands_CreateDebugEvents(t *testing.T) { + ctx := authz.NewMockContextWithPermissions("instance1", "org1", "user1", nil) + type fields struct { + eventstore func(*testing.T) *eventstore.Eventstore + } + type args struct { + dbe *DebugEvents + } + tests := []struct { + name string + fields fields + args args + want *domain.ObjectDetails + wantErr error + }{ + { + name: "filter error", + fields: fields{ + eventstore: expectEventstore( + expectFilterError(io.ErrClosedPipe), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventAdded{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("a"), + }, + }, + }}, + wantErr: io.ErrClosedPipe, + }, + { + name: "already exists", + fields: fields{ + eventstore: expectEventstore( + expectFilter( + eventFromEventPusher( + debug_events.NewAddedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("a"), + ), + ), + ), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventAdded{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("a"), + }, + }, + }}, + wantErr: zerrors.ThrowAlreadyExists(nil, "COMMAND-Aex6j", "debug aggregate already exists"), + }, + { + name: "double added event, already exists", + fields: fields{ + eventstore: expectEventstore( + expectFilter(), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventAdded{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("a"), + }, + DebugEventAdded{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("a"), + }, + }, + }}, + wantErr: zerrors.ThrowAlreadyExists(nil, "COMMAND-Aex6j", "debug aggregate already exists"), + }, + { + name: "changed event, not found", + fields: fields{ + eventstore: expectEventstore( + expectFilter(), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventChanged{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("a"), + }, + }, + }}, + wantErr: zerrors.ThrowNotFound(nil, "COMMAND-Thie6", "debug aggregate not found"), + }, + { + name: "removed event, not found", + fields: fields{ + eventstore: expectEventstore( + expectFilter(), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventRemoved{ + ProjectionSleep: time.Millisecond, + }, + }, + }}, + wantErr: zerrors.ThrowNotFound(nil, "COMMAND-Ohna9", "debug aggregate not found"), + }, + { + name: "changed after removed event, not found", + fields: fields{ + eventstore: expectEventstore( + expectFilter( + eventFromEventPusher( + debug_events.NewAddedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("a"), + ), + ), + ), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventRemoved{ + ProjectionSleep: time.Millisecond, + }, + DebugEventChanged{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("a"), + }, + }, + }}, + wantErr: zerrors.ThrowNotFound(nil, "COMMAND-Thie6", "debug aggregate not found"), + }, + { + name: "double removed event, not found", + fields: fields{ + eventstore: expectEventstore( + expectFilter( + eventFromEventPusher( + debug_events.NewAddedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("a"), + ), + ), + ), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventRemoved{ + ProjectionSleep: time.Millisecond, + }, + DebugEventRemoved{ + ProjectionSleep: time.Millisecond, + }, + }, + }}, + wantErr: zerrors.ThrowNotFound(nil, "COMMAND-Ohna9", "debug aggregate not found"), + }, + { + name: "added, ok", + fields: fields{ + eventstore: expectEventstore( + expectFilter(), + expectPush( + debug_events.NewAddedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("a"), + ), + ), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventAdded{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("a"), + }, + }, + }}, + want: &domain.ObjectDetails{ + ResourceOwner: "instance1", + }, + }, + { + name: "changed, ok", + fields: fields{ + eventstore: expectEventstore( + expectFilter( + eventFromEventPusher( + debug_events.NewAddedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("a"), + ), + ), + ), + expectPush( + debug_events.NewChangedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("b"), + ), + ), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventChanged{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("b"), + }, + }, + }}, + want: &domain.ObjectDetails{ + ResourceOwner: "instance1", + }, + }, + { + name: "removed, ok", + fields: fields{ + eventstore: expectEventstore( + expectFilter( + eventFromEventPusher( + debug_events.NewAddedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("a"), + ), + ), + ), + expectPush( + debug_events.NewRemovedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, + ), + ), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventRemoved{ + ProjectionSleep: time.Millisecond, + }, + }, + }}, + want: &domain.ObjectDetails{ + ResourceOwner: "instance1", + }, + }, + { + name: "added, changed, changed, removed ok", + fields: fields{ + eventstore: expectEventstore( + expectFilter(), + expectPush( + debug_events.NewAddedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("a"), + ), + debug_events.NewChangedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("b"), + ), + debug_events.NewChangedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, gu.Ptr("c"), + ), + debug_events.NewRemovedEvent( + ctx, debug_events.NewAggregate("dbg1", "instance1"), + time.Millisecond, + ), + ), + ), + }, + args: args{&DebugEvents{ + AggregateID: "dbg1", + Events: []DebugEvent{ + DebugEventAdded{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("a"), + }, + DebugEventChanged{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("b"), + }, + DebugEventChanged{ + ProjectionSleep: time.Millisecond, + Blob: gu.Ptr("c"), + }, + DebugEventRemoved{ + ProjectionSleep: time.Millisecond, + }, + }, + }}, + want: &domain.ObjectDetails{ + ResourceOwner: "instance1", + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := &Commands{ + eventstore: tt.fields.eventstore(t), + } + got, err := c.CreateDebugEvents(ctx, tt.args.dbe) + require.ErrorIs(t, err, tt.wantErr) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/internal/domain/debug_events.go b/internal/domain/debug_events.go new file mode 100644 index 000000000..627c4e54b --- /dev/null +++ b/internal/domain/debug_events.go @@ -0,0 +1,14 @@ +package domain + +type DebugEventsState int + +const ( + DebugEventsStateUnspecified DebugEventsState = iota + DebugEventsStateInitial + DebugEventsStateChanged + DebugEventsStateRemoved +) + +func (state DebugEventsState) Exists() bool { + return state == DebugEventsStateInitial || state == DebugEventsStateChanged +} diff --git a/internal/eventstore/handler/v2/statement.go b/internal/eventstore/handler/v2/statement.go index 273789f01..4bc660d9f 100644 --- a/internal/eventstore/handler/v2/statement.go +++ b/internal/eventstore/handler/v2/statement.go @@ -338,6 +338,21 @@ func NewNoOpStatement(event eventstore.Event) *Statement { return NewStatement(event, nil) } +func NewSleepStatement(event eventstore.Event, d time.Duration, opts ...execOption) *Statement { + return NewStatement( + event, + exec( + execConfig{ + args: []any{float64(d) / float64(time.Second)}, + }, + func(_ execConfig) string { + return "SELECT pg_sleep($1);" + }, + opts, + ), + ) +} + func NewMultiStatement(event eventstore.Event, opts ...func(eventstore.Event) Exec) *Statement { if len(opts) == 0 { return NewNoOpStatement(event) @@ -385,6 +400,12 @@ func AddCopyStatement(conflict, from, to []Column, conditions []NamespacedCondit } } +func AddSleepStatement(d time.Duration, opts ...execOption) func(eventstore.Event) Exec { + return func(event eventstore.Event) Exec { + return NewSleepStatement(event, d, opts...).Execute + } +} + func NewArrayAppendCol(column string, value interface{}) Column { return Column{ Name: column, diff --git a/internal/query/debug_events.go b/internal/query/debug_events.go new file mode 100644 index 000000000..cc3e6e6cd --- /dev/null +++ b/internal/query/debug_events.go @@ -0,0 +1,106 @@ +package query + +import ( + "context" + "database/sql" + _ "embed" + "errors" + + "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/domain" + "github.com/zitadel/zitadel/internal/eventstore/handler/v2" + "github.com/zitadel/zitadel/internal/query/projection" + "github.com/zitadel/zitadel/internal/telemetry/tracing" + "github.com/zitadel/zitadel/internal/zerrors" +) + +type DebugEventState struct { + domain.ObjectDetails + Blob string +} + +var ( + //go:embed debug_events_state_by_id.sql + debugEventsStateByIdQuery string + //go:embed debug_events_states.sql + debugEventsStatesQuery string +) + +func (q *Queries) GetDebugEventsStateByID(ctx context.Context, id string, triggerBulk bool) (_ *DebugEventState, err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + ctx, err = triggerDebugEventsProjection(ctx, triggerBulk) + if err != nil { + return nil, err + } + + dst := new(DebugEventState) + err = q.client.QueryRowContext(ctx, + func(row *sql.Row) error { + return row.Scan( + &dst.ID, + &dst.CreationDate, + &dst.EventDate, + &dst.ResourceOwner, + &dst.Sequence, + &dst.Blob, + ) + }, + debugEventsStateByIdQuery, + authz.GetInstance(ctx).InstanceID(), + id, + ) + if errors.Is(err, sql.ErrNoRows) { + return nil, zerrors.ThrowNotFound(err, "QUERY-Eeth5", "debug event state not found") + } + if err != nil { + return nil, zerrors.ThrowInternal(err, "QUERY-oe0Ae", "Errors.Internal") + } + return dst, err +} + +func (q *Queries) ListDebugEventsStates(ctx context.Context, triggerBulk bool) (out []DebugEventState, err error) { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + ctx, err = triggerDebugEventsProjection(ctx, triggerBulk) + if err != nil { + return nil, err + } + + err = q.client.QueryContext(ctx, + func(rows *sql.Rows) error { + for rows.Next() { + var dst DebugEventState + err := rows.Scan( + &dst.ID, + &dst.CreationDate, + &dst.EventDate, + &dst.ResourceOwner, + &dst.Sequence, + &dst.Blob, + ) + if err != nil { + return err + } + out = append(out, dst) + } + return rows.Err() + }, + debugEventsStatesQuery, + authz.GetInstance(ctx).InstanceID(), + ) + + if err != nil { + return nil, zerrors.ThrowInternal(err, "QUERY-nooZ2", "Errors.Internal") + } + return out, nil +} + +func triggerDebugEventsProjection(ctx context.Context, trigger bool) (_ context.Context, err error) { + if trigger { + ctx, span := tracing.NewSpan(ctx) + defer func() { span.EndWithError(err) }() + return projection.DebugEventsProjection.Trigger(ctx, handler.WithAwaitRunning()) + } + return ctx, nil +} diff --git a/internal/query/debug_events_state_by_id.sql b/internal/query/debug_events_state_by_id.sql new file mode 100644 index 000000000..a91f38f9d --- /dev/null +++ b/internal/query/debug_events_state_by_id.sql @@ -0,0 +1,5 @@ +select id, creation_date, change_date, resource_owner, sequence, blob +from projections.debug_events +where instance_id = $1 +and id = $2 +limit 1; diff --git a/internal/query/debug_events_states.sql b/internal/query/debug_events_states.sql new file mode 100644 index 000000000..ec3e66bb8 --- /dev/null +++ b/internal/query/debug_events_states.sql @@ -0,0 +1,4 @@ +select id, creation_date, change_date, resource_owner, sequence, blob +from projections.debug_events +where instance_id = $1 +order by creation_date asc; diff --git a/internal/query/projection/debug_events.go b/internal/query/projection/debug_events.go new file mode 100644 index 000000000..687bcc2c9 --- /dev/null +++ b/internal/query/projection/debug_events.go @@ -0,0 +1,146 @@ +package projection + +import ( + "context" + + "github.com/muhlemmer/gu" + + "github.com/zitadel/zitadel/internal/eventstore" + old_handler "github.com/zitadel/zitadel/internal/eventstore/handler" + "github.com/zitadel/zitadel/internal/eventstore/handler/v2" + "github.com/zitadel/zitadel/internal/repository/debug_events" + "github.com/zitadel/zitadel/internal/repository/instance" + "github.com/zitadel/zitadel/internal/zerrors" +) + +const ( + DebugEventsProjectionTable = "projections.debug_events" + + DebugEventsColumnID = "id" + DebugEventsColumnCreationDate = "creation_date" + DebugEventsColumnChangeDate = "change_date" + DebugEventsColumnResourceOwner = "resource_owner" + DebugEventsColumnInstanceID = "instance_id" + DebugEventsColumnSequence = "sequence" + DebugEventsColumnBlob = "blob" +) + +type debugEventsProjection struct{} + +func (*debugEventsProjection) Name() string { + return DebugEventsProjectionTable +} + +func newDebugEventsProjection(ctx context.Context, config handler.Config) *handler.Handler { + return handler.NewHandler(ctx, &config, new(debugEventsProjection)) +} + +// Init implements [handler.initializer] +func (p *debugEventsProjection) Init() *old_handler.Check { + return handler.NewTableCheck( + handler.NewTable([]*handler.InitColumn{ + handler.NewColumn(DebugEventsColumnID, handler.ColumnTypeText), + handler.NewColumn(DebugEventsColumnCreationDate, handler.ColumnTypeTimestamp), + handler.NewColumn(DebugEventsColumnChangeDate, handler.ColumnTypeTimestamp), + handler.NewColumn(DebugEventsColumnResourceOwner, handler.ColumnTypeText), + handler.NewColumn(DebugEventsColumnInstanceID, handler.ColumnTypeText), + handler.NewColumn(DebugEventsColumnSequence, handler.ColumnTypeInt64), + handler.NewColumn(DebugEventsColumnBlob, handler.ColumnTypeText), + }, + handler.NewPrimaryKey(DebugEventsColumnInstanceID, DebugEventsColumnID), + ), + ) +} + +func (p *debugEventsProjection) Reducers() []handler.AggregateReducer { + return []handler.AggregateReducer{ + { + Aggregate: debug_events.AggregateType, + EventReducers: []handler.EventReducer{ + { + Event: debug_events.AddedEventType, + Reduce: p.reduceDebugEventAdded, + }, + { + Event: debug_events.ChangedEventType, + Reduce: p.reduceDebugEventChanged, + }, + { + Event: debug_events.RemovedEventType, + Reduce: p.reduceDebugEventRemoved, + }, + }, + }, + { + Aggregate: instance.AggregateType, + EventReducers: []handler.EventReducer{ + { + Event: instance.InstanceRemovedEventType, + Reduce: reduceInstanceRemovedHelper(DebugEventsColumnInstanceID), + }, + }, + }, + } +} + +func (p *debugEventsProjection) reduceDebugEventAdded(event eventstore.Event) (*handler.Statement, error) { + e, ok := event.(*debug_events.AddedEvent) + if !ok { + return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-uYq4r", "reduce.wrong.event.type %s", debug_events.AddedEventType) + } + return handler.NewMultiStatement( + e, + handler.AddSleepStatement(e.ProjectionSleep), + handler.AddCreateStatement([]handler.Column{ + handler.NewCol(DebugEventsColumnID, e.Aggregate().ID), + handler.NewCol(DebugEventsColumnCreationDate, e.CreationDate()), + handler.NewCol(DebugEventsColumnChangeDate, e.CreationDate()), + handler.NewCol(DebugEventsColumnResourceOwner, e.Aggregate().ResourceOwner), + handler.NewCol(DebugEventsColumnInstanceID, e.Aggregate().InstanceID), + handler.NewCol(DebugEventsColumnSequence, e.Sequence()), + handler.NewCol(DebugEventsColumnBlob, gu.Value(e.Blob)), + }), + ), nil +} + +func (p *debugEventsProjection) reduceDebugEventChanged(event eventstore.Event) (*handler.Statement, error) { + e, ok := event.(*debug_events.ChangedEvent) + if !ok { + return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-Bg8oM", "reduce.wrong.event.type %s", debug_events.ChangedEventType) + } + updateCols := []handler.Column{ + handler.NewCol(DebugEventsColumnChangeDate, e.CreationDate()), + handler.NewCol(DebugEventsColumnSequence, e.Sequence()), + } + if e.Blob != nil { + updateCols = append(updateCols, + handler.NewCol(DebugEventsColumnBlob, *e.Blob), + ) + } + + return handler.NewMultiStatement( + e, + handler.AddSleepStatement(e.ProjectionSleep), + handler.AddUpdateStatement(updateCols, + []handler.Condition{ + handler.NewCond(DebugEventsColumnID, e.Aggregate().ID), + handler.NewCond(DebugEventsColumnInstanceID, e.Aggregate().InstanceID), + }), + ), nil +} + +func (p *debugEventsProjection) reduceDebugEventRemoved(event eventstore.Event) (*handler.Statement, error) { + e, ok := event.(*debug_events.RemovedEvent) + if !ok { + return nil, zerrors.ThrowInvalidArgumentf(nil, "PROJE-DgMSg", "reduce.wrong.event.type %s", debug_events.RemovedEventType) + } + return handler.NewMultiStatement( + e, + handler.AddSleepStatement(e.ProjectionSleep), + handler.AddDeleteStatement( + []handler.Condition{ + handler.NewCond(DebugEventsColumnID, e.Aggregate().ID), + handler.NewCond(DebugEventsColumnInstanceID, e.Aggregate().InstanceID), + }), + ), nil +} diff --git a/internal/query/projection/projection.go b/internal/query/projection/projection.go index 0151a9953..c4660c6c3 100644 --- a/internal/query/projection/projection.go +++ b/internal/query/projection/projection.go @@ -79,6 +79,7 @@ var ( ExecutionProjection *handler.Handler UserSchemaProjection *handler.Handler WebKeyProjection *handler.Handler + DebugEventsProjection *handler.Handler ProjectGrantFields *handler.FieldHandler OrgDomainVerifiedFields *handler.FieldHandler @@ -165,6 +166,7 @@ func Create(ctx context.Context, sqlClient *database.DB, es handler.EventStore, ExecutionProjection = newExecutionProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["executions"])) UserSchemaProjection = newUserSchemaProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["user_schemas"])) WebKeyProjection = newWebKeyProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["web_keys"])) + DebugEventsProjection = newDebugEventsProjection(ctx, applyCustomConfig(projectionConfig, config.Customizations["debug_events"])) ProjectGrantFields = newFillProjectGrantFields(applyCustomConfig(projectionConfig, config.Customizations[fieldsProjectGrant])) OrgDomainVerifiedFields = newFillOrgDomainVerifiedFields(applyCustomConfig(projectionConfig, config.Customizations[fieldsOrgDomainVerified])) @@ -295,5 +297,6 @@ func newProjectionsList() { ExecutionProjection, UserSchemaProjection, WebKeyProjection, + DebugEventsProjection, } } diff --git a/internal/repository/debug_events/aggregate.go b/internal/repository/debug_events/aggregate.go new file mode 100644 index 000000000..f2abd17b5 --- /dev/null +++ b/internal/repository/debug_events/aggregate.go @@ -0,0 +1,27 @@ +package debug_events + +import ( + "github.com/zitadel/zitadel/internal/eventstore" +) + +const ( + eventTypePrefix = eventstore.EventType("debug.") +) + +const ( + AggregateType = "debug" + AggregateVersion = "v1" +) + +type Aggregate struct { + eventstore.Aggregate +} + +func NewAggregate(id, resourceOwner string) *eventstore.Aggregate { + return &eventstore.Aggregate{ + Type: AggregateType, + Version: AggregateVersion, + ID: id, + ResourceOwner: resourceOwner, + } +} diff --git a/internal/repository/debug_events/debug.go b/internal/repository/debug_events/debug.go new file mode 100644 index 000000000..f56d995d6 --- /dev/null +++ b/internal/repository/debug_events/debug.go @@ -0,0 +1,125 @@ +package debug_events + +import ( + "context" + "time" + + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/zerrors" +) + +const ( + AddedEventType = eventTypePrefix + "added" + ChangedEventType = eventTypePrefix + "changed" + RemovedEventType = eventTypePrefix + "removed" +) + +type AddedEvent struct { + eventstore.BaseEvent `json:"-"` + ProjectionSleep time.Duration `json:"projectionSleep,omitempty"` + Blob *string `json:"blob,omitempty"` +} + +func (e *AddedEvent) Payload() interface{} { + return e +} + +func (e *AddedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { + return nil +} + +func NewAddedEvent(ctx context.Context, aggregate *eventstore.Aggregate, projectionSleep time.Duration, blob *string) *AddedEvent { + return &AddedEvent{ + BaseEvent: *eventstore.NewBaseEventForPush( + ctx, + aggregate, + AddedEventType, + ), + Blob: blob, + ProjectionSleep: projectionSleep, + } +} + +func DebugAddedEventMapper(event eventstore.Event) (eventstore.Event, error) { + debugAdded := &AddedEvent{ + BaseEvent: *eventstore.BaseEventFromRepo(event), + } + err := event.Unmarshal(debugAdded) + if err != nil { + return nil, zerrors.ThrowInternal(err, "ORG-Bren2", "unable to unmarshal debug added") + } + + return debugAdded, nil +} + +type ChangedEvent struct { + eventstore.BaseEvent `json:"-"` + ProjectionSleep time.Duration `json:"projectionSleep,omitempty"` + Blob *string `json:"blob,omitempty"` +} + +func (e *ChangedEvent) Payload() interface{} { + return e +} + +func (e *ChangedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { + return nil +} + +func NewChangedEvent(ctx context.Context, aggregate *eventstore.Aggregate, projectionSleep time.Duration, blob *string) *ChangedEvent { + return &ChangedEvent{ + BaseEvent: *eventstore.NewBaseEventForPush( + ctx, + aggregate, + ChangedEventType, + ), + ProjectionSleep: projectionSleep, + Blob: blob, + } +} + +func DebugChangedEventMapper(event eventstore.Event) (eventstore.Event, error) { + debugChanged := &ChangedEvent{ + BaseEvent: *eventstore.BaseEventFromRepo(event), + } + err := event.Unmarshal(debugChanged) + if err != nil { + return nil, zerrors.ThrowInternal(err, "ORG-Bren2", "unable to unmarshal debug added") + } + + return debugChanged, nil +} + +type RemovedEvent struct { + eventstore.BaseEvent `json:"-"` + ProjectionSleep time.Duration `json:"projectionSleep,omitempty"` +} + +func (e *RemovedEvent) Payload() interface{} { + return nil +} + +func (e *RemovedEvent) UniqueConstraints() []*eventstore.UniqueConstraint { + return nil +} + +func NewRemovedEvent(ctx context.Context, aggregate *eventstore.Aggregate, projectionSleep time.Duration) *RemovedEvent { + return &RemovedEvent{ + BaseEvent: *eventstore.NewBaseEventForPush( + ctx, + aggregate, + RemovedEventType, + ), + ProjectionSleep: projectionSleep, + } +} + +func DebugRemovedEventMapper(event eventstore.Event) (eventstore.Event, error) { + return &RemovedEvent{ + BaseEvent: *eventstore.BaseEventFromRepo(event), + }, nil +} + +func AggregateFromWriteModel(ctx context.Context, wm *eventstore.WriteModel) *eventstore.Aggregate { + return eventstore.AggregateFromWriteModelCtx(ctx, wm, AggregateType, AggregateVersion) +} diff --git a/internal/repository/debug_events/eventstore.go b/internal/repository/debug_events/eventstore.go new file mode 100644 index 000000000..f774d2140 --- /dev/null +++ b/internal/repository/debug_events/eventstore.go @@ -0,0 +1,11 @@ +package debug_events + +import ( + "github.com/zitadel/zitadel/internal/eventstore" +) + +func init() { + eventstore.RegisterFilterEventMapper(AggregateType, AddedEventType, DebugAddedEventMapper) + eventstore.RegisterFilterEventMapper(AggregateType, ChangedEventType, DebugChangedEventMapper) + eventstore.RegisterFilterEventMapper(AggregateType, RemovedEventType, DebugRemovedEventMapper) +} diff --git a/proto/zitadel/resources/debug_events/v3alpha/debug_events_service.proto b/proto/zitadel/resources/debug_events/v3alpha/debug_events_service.proto new file mode 100644 index 000000000..6a5990f78 --- /dev/null +++ b/proto/zitadel/resources/debug_events/v3alpha/debug_events_service.proto @@ -0,0 +1,223 @@ +syntax = "proto3"; + +package zitadel.resources.debug_events.v3alpha; + +import "google/api/annotations.proto"; +import "google/api/field_behavior.proto"; +import "protoc-gen-openapiv2/options/annotations.proto"; +import "validate/validate.proto"; + +import "zitadel/protoc_gen_zitadel/v2/options.proto"; +import "zitadel/object/v3alpha/object.proto"; +import "zitadel/resources/object/v3alpha/object.proto"; +import "zitadel/resources/debug_events/v3alpha/event.proto"; +import "zitadel/resources/debug_events/v3alpha/state.proto"; + +option go_package = "github.com/zitadel/zitadel/pkg/grpc/resources/debug_events/v3alpha;debug_events"; + +option (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_swagger) = { + info: { + title: "Debug Service"; + version: "3.0-preview"; + description: "This API is intended to push specific debug payload through ZITADEL's storage system."; + contact:{ + name: "ZITADEL" + url: "https://zitadel.com" + email: "hi@zitadel.com" + } + license: { + name: "Apache 2.0", + url: "https://github.com/zitadel/zitadel/blob/main/LICENSE"; + }; + }; + schemes: HTTPS; + schemes: HTTP; + + consumes: "application/json"; + produces: "application/json"; + + consumes: "application/grpc"; + produces: "application/grpc"; + + consumes: "application/grpc-web+proto"; + produces: "application/grpc-web+proto"; + + host: "$CUSTOM-DOMAIN"; + base_path: "/resources/v3alpha/debug"; + + external_docs: { + description: "Detailed information about ZITADEL", + url: "https://zitadel.com/docs" + } + security_definitions: { + security: { + key: "OAuth2"; + value: { + type: TYPE_OAUTH2; + flow: FLOW_ACCESS_CODE; + authorization_url: "$CUSTOM-DOMAIN/oauth/v2/authorize"; + token_url: "$CUSTOM-DOMAIN/oauth/v2/token"; + scopes: { + scope: { + key: "openid"; + value: "openid"; + } + scope: { + key: "urn:zitadel:iam:org:project:id:zitadel:aud"; + value: "urn:zitadel:iam:org:project:id:zitadel:aud"; + } + } + } + } + } + security: { + security_requirement: { + key: "OAuth2"; + value: { + scope: "openid"; + scope: "urn:zitadel:iam:org:project:id:zitadel:aud"; + } + } + } + responses: { + key: "403"; + value: { + description: "Returned when the user does not have permission to access the resource."; + schema: { + json_schema: { + ref: "#/definitions/rpcStatus"; + } + } + } + } + responses: { + key: "404"; + value: { + description: "Returned when the resource does not exist."; + schema: { + json_schema: { + ref: "#/definitions/rpcStatus"; + } + } + } + } +}; + +service ZITADELDebugEvents { + rpc CreateDebugEvents(CreateDebugEventsRequest) returns (CreateDebugEventsResponse) { + option (google.api.http) = { + post: "/" + body: "events" + }; + + option (zitadel.protoc_gen_zitadel.v2.options) = { + auth_option: { + permission: "iam.debug.write" + } + http_response: { + success_code: 201 + } + }; + + option (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_operation) = { + summary: "Create a set of debug events."; + description: "Create a set of debug events which will be pushed to the eventstore and reduced to the projection." + responses: { + key: "200" + value: { + description: "OK"; + } + }; + }; + } + + rpc GetDebugEventsStateById(GetDebugEventsStateByIdRequest) returns (GetDebugEventsStateByIdResponse) { + option (google.api.http) = { + get: "/v3alpha/debug_events/{id}" + }; + + option (zitadel.protoc_gen_zitadel.v2.options) = { + auth_option: { + permission: "iam.debug.read" + } + }; + + option (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_operation) = { + responses: { + key: "200" + value: { + description: "Debug events state successfully retrieved"; + } + }; + }; + } + + rpc ListDebugEventsStates(ListDebugEventsStatesRequest) returns (ListDebugEventsStatesResponse) { + option (google.api.http) = { + get: "/v3alpha/debug_events" + }; + + option (zitadel.protoc_gen_zitadel.v2.options) = { + auth_option: { + permission: "iam.debug.read" + } + }; + + option (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_operation) = { + responses: { + key: "200" + value: { + description: "Debug events states successfully retrieved"; + } + }; + }; + } +} + +message CreateDebugEventsRequest { + optional zitadel.object.v3alpha.Instance instance = 1 [ + (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { + default: "\"domain from HOST or :authority header\"" + } + ]; + // unique identifier for the aggregate we want to push events to. + string aggregate_id = 2 [ + (google.api.field_behavior) = REQUIRED, + (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { + min_length: 1, + max_length: 200, + example: "\"69629026806489455\""; + } + ]; + repeated Event events = 3; +} + +message CreateDebugEventsResponse { + zitadel.resources.object.v3alpha.Details details = 1; +} + +message GetDebugEventsStateByIdRequest { + // unique identifier of the aggregate. + string id = 1 [ + (validate.rules).string = {min_len: 1, max_len: 200}, + (google.api.field_behavior) = REQUIRED, + (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { + min_length: 1, + max_length: 200, + example: "\"69629026806489455\""; + } + ]; + bool trigger_bulk = 2; +} + +message GetDebugEventsStateByIdResponse { + State state = 1; +} + +message ListDebugEventsStatesRequest { + bool trigger_bulk = 1; +} + +message ListDebugEventsStatesResponse { + repeated State states = 1; +} \ No newline at end of file diff --git a/proto/zitadel/resources/debug_events/v3alpha/event.proto b/proto/zitadel/resources/debug_events/v3alpha/event.proto new file mode 100644 index 000000000..096c9918b --- /dev/null +++ b/proto/zitadel/resources/debug_events/v3alpha/event.proto @@ -0,0 +1,47 @@ +syntax = "proto3"; + +package zitadel.resources.debug_events.v3alpha; + +import "google/protobuf/duration.proto"; +import "protoc-gen-openapiv2/options/annotations.proto"; + +option go_package = "github.com/zitadel/zitadel/pkg/grpc/resources/debug_events/v3alpha;debug_events"; + +message Event { + oneof event { + AddedEvent add = 1; + ChangedEvent change = 2; + RemovedEvent remove = 3; + } +} + +message AddedEvent { + // issues a pg_sleep command in the projection reducer, simulating a slow query. + google.protobuf.Duration projection_sleep = 1 [ + (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { + example: "\"5s\""; + } + ]; + // optional text that can be set as a state. + optional string blob = 2; +} + +message ChangedEvent { + // issues a pg_sleep command in the projection reducer, simulating a slow query. + google.protobuf.Duration projection_sleep = 1 [ + (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { + example: "\"5s\""; + } + ]; + // optional text that can be set as a state. + optional string blob = 2; +} + +message RemovedEvent { + // issues a pg_sleep command in the projection reducer, simulating a slow query. + google.protobuf.Duration projection_sleep = 1 [ + (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { + example: "\"5s\""; + } + ]; +} \ No newline at end of file diff --git a/proto/zitadel/resources/debug_events/v3alpha/state.proto b/proto/zitadel/resources/debug_events/v3alpha/state.proto new file mode 100644 index 000000000..5cdf60d74 --- /dev/null +++ b/proto/zitadel/resources/debug_events/v3alpha/state.proto @@ -0,0 +1,14 @@ +syntax = "proto3"; + +package zitadel.resources.debug_events.v3alpha; + +import "zitadel/resources/object/v3alpha/object.proto"; + +option go_package = "github.com/zitadel/zitadel/pkg/grpc/resources/debug_events/v3alpha;debug_events"; + +message State { + // Details provide some base information (such as the last change date) of the schema. + zitadel.resources.object.v3alpha.Details details = 1; + + string blob = 2; +} \ No newline at end of file