stop tracking events

internal/command/command.go

@@ -5,6 +5,8 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/zitadel/zitadel/internal/repository/milestone"
+
 	"github.com/zitadel/zitadel/internal/api/authz"
 	api_http "github.com/zitadel/zitadel/internal/api/http"
 	"github.com/zitadel/zitadel/internal/command/preparation"
@@ -124,6 +126,7 @@ func StartCommands(
 	quota.RegisterEventMappers(repo.eventstore)
 	session.RegisterEventMappers(repo.eventstore)
 	idpintent.RegisterEventMappers(repo.eventstore)
+	milestone.RegisterEventMappers(repo.eventstore)
 
 	repo.userPasswordAlg = crypto.NewBCrypt(defaults.SecretGenerators.PasswordSaltCost)
 	repo.machineKeySize = int(defaults.SecretGenerators.MachineKeySize)

internal/command/telemetry.go

@@ -3,25 +3,23 @@ package command
 import (
 	"context"
 
-	"github.com/zitadel/zitadel/internal/repository/milestone"
-
 	"github.com/zitadel/zitadel/internal/eventstore"
+
+	"github.com/zitadel/zitadel/internal/repository/milestone"
 )
 
-// ReportTelemetryUsage writes one or many *telemetry.PushDueEvent directly to the eventstore
-func (c *Commands) ReportTelemetryUsage(ctx context.Context, dueEvent ...*milestone.ReachedEvent) error {
-	cmds := make([]eventstore.Command, len(dueEvent))
-	for idx, notification := range dueEvent {
-		cmds[idx] = notification
+// ReportMilestoneReached writes each *milestone.ReachedEvent directly to the event store
+func (c *Commands) ReportMilestoneReached(ctx context.Context, triggeringEvent eventstore.Event, customContext interface{}) error {
+	aggregateId, err := c.idGenerator.Next()
+	if err != nil {
+		return err
 	}
-	_, err := c.eventstore.Push(ctx, cmds...)
+	_, err = c.eventstore.Push(ctx, milestone.NewReachedEvent(ctx, aggregateId, triggeringEvent, customContext))
 	return err
 }
 
-func (c *Commands) TelemetryPushed(ctx context.Context, dueEvent *milestone.ReachedEvent, endpoints []string) error {
-	_, err := c.eventstore.Push(
-		ctx,
-		milestone.NewPushedEvent(ctx, dueEvent, endpoints),
-	)
+// ReportMilestonePushed defers a milestone.PushedEvent for each *milestone.ReachedEvent and writes it directly to the event store.
+func (c *Commands) ReportMilestonePushed(ctx context.Context, endpoints []string, reachedEvent *milestone.ReachedEvent) error {
+	_, err := c.eventstore.Push(ctx, milestone.NewPushedEvent(ctx, reachedEvent, endpoints))
 	return err
 }

internal/notification/handlers/telemetrypusher.go

@@ -27,12 +27,12 @@ type TelemetryPusherConfig struct {
 }
 
 type telemetryPusher struct {
-	cfg TelemetryPusherConfig
 	crdb.StatementHandler
 	commands                       *command.Commands
 	queries                        *NotificationQueries
 	metricSuccessfulDeliveriesJSON string
 	metricFailedDeliveriesJSON     string
+	endpoints                      []string
 }
 
 func NewTelemetryPusher(
@@ -46,8 +46,11 @@ func NewTelemetryPusher(
 ) *telemetryPusher {
 	p := new(telemetryPusher)
 	handlerCfg.ProjectionName = TelemetryProjectionTable
-	handlerCfg.Reducers = p.reducers()
-	p.cfg = telemetryCfg
+	handlerCfg.Reducers = []handler.AggregateReducer{{}}
+	if telemetryCfg.Enabled {
+		handlerCfg.Reducers = p.reducers()
+	}
+	p.endpoints = telemetryCfg.Endpoints
 	p.StatementHandler = crdb.NewStatementHandler(ctx, handlerCfg)
 	p.commands = commands
 	p.queries = queries
@@ -58,23 +61,20 @@ func NewTelemetryPusher(
 }
 
 func (t *telemetryPusher) reducers() []handler.AggregateReducer {
-	if !t.cfg.Enabled {
-		return nil
-	}
 	return []handler.AggregateReducer{
 		{
 			Aggregate: milestone.AggregateType,
 			EventRedusers: []handler.EventReducer{
 				{
 					Event:  milestone.ReachedEventType,
-					Reduce: t.reduceTelemetryPushDue,
+					Reduce: t.reduceMilestoneReached,
 				},
 			},
 		},
 	}
 }
 
-func (t *telemetryPusher) reduceTelemetryPushDue(event eventstore.Event) (*handler.Statement, error) {
+func (t *telemetryPusher) reduceMilestoneReached(event eventstore.Event) (*handler.Statement, error) {
 	e, ok := event.(*milestone.ReachedEvent)
 	if !ok {
 		return nil, errors.ThrowInvalidArgumentf(nil, "HANDL-UjA3E", "reduce.wrong.event.type %s", milestone.ReachedEventType)
@@ -87,7 +87,7 @@ func (t *telemetryPusher) reduceTelemetryPushDue(event eventstore.Event) (*handl
 	if alreadyHandled {
 		return crdb.NewNoOpStatement(e), nil
 	}
-	for _, endpoint := range t.cfg.Endpoints {
+	for _, endpoint := range t.endpoints {
 		if err = types.SendJSON(
 			ctx,
 			webhook.Config{
@@ -105,7 +105,7 @@ func (t *telemetryPusher) reduceTelemetryPushDue(event eventstore.Event) (*handl
 		}
 	}
 
-	err = t.commands.TelemetryPushed(ctx, e, t.cfg.Endpoints)
+	err = t.commands.ReportMilestonePushed(ctx, t.endpoints, e)
 	if err != nil {
 		return nil, err
 	}

internal/notification/handlers/usernotifier.go

@@ -2,13 +2,8 @@ package handlers
 
 import (
 	"context"
-	"fmt"
 	"time"
 
-	"github.com/zitadel/zitadel/internal/repository/project"
-
-	"github.com/zitadel/zitadel/internal/repository/instance"
-
 	"github.com/zitadel/zitadel/internal/command"
 	"github.com/zitadel/zitadel/internal/crypto"
 	"github.com/zitadel/zitadel/internal/domain"
@@ -111,67 +106,11 @@ func (u *userNotifier) reducers() []handler.AggregateReducer {
 					Event:  user.HumanPasswordChangedType,
 					Reduce: u.reducePasswordChanged,
 				},
-				{
-					Event:  user.UserTokenAddedType,
-					Reduce: u.reduceUserTokenAdded,
-				},
-			},
-		},
-		{
-			Aggregate: instance.AggregateType,
-			EventRedusers: []handler.EventReducer{
-				{
-					Event:  instance.InstanceAddedEventType,
-					Reduce: u.reduceInstanceAdded,
-				},
-				{
-					Event:  instance.InstanceRemovedEventType,
-					Reduce: u.reduceInstanceRemoved,
-				},
-			},
-		},
-		{
-			Aggregate: project.AggregateType,
-			EventRedusers: []handler.EventReducer{
-				{
-					Event:  project.ProjectAddedType,
-					Reduce: u.reduceProjectAdded,
-				},
-				{
-					Event:  project.ApplicationAddedType,
-					Reduce: u.reduceApplicationAdded,
-				},
 			},
 		},
 	}
 }
 
-func (u *userNotifier) reduceInstanceAdded(event eventstore.Event) (*handler.Statement, error) {
-	fmt.Println("reduceInstanceAdded")
-	return crdb.NewNoOpStatement(event), nil
-}
-
-func (u *userNotifier) reduceProjectAdded(event eventstore.Event) (*handler.Statement, error) {
-	// ignore instance.ProjectSetEventType
-	fmt.Println("reduceProjectAdded")
-	return crdb.NewNoOpStatement(event), nil
-}
-
-func (u *userNotifier) reduceApplicationAdded(event eventstore.Event) (*handler.Statement, error) {
-	fmt.Println("reduceApplicationAdded")
-	return crdb.NewNoOpStatement(event), nil
-}
-
-func (u *userNotifier) reduceUserTokenAdded(event eventstore.Event) (*handler.Statement, error) {
-	fmt.Println("reduceUserTokenAdded")
-	return crdb.NewNoOpStatement(event), nil
-}
-
-func (u *userNotifier) reduceInstanceRemoved(event eventstore.Event) (*handler.Statement, error) {
-	fmt.Println("reduceInstanceRemoved")
-	return crdb.NewNoOpStatement(event), nil
-}
-
 func (u *userNotifier) reduceInitCodeAdded(event eventstore.Event) (*handler.Statement, error) {
 	e, ok := event.(*user.HumanInitialCodeAddedEvent)
 	if !ok {

internal/query/projection/milestones.go

@@ -0,0 +1,169 @@
+package projection
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/zitadel/zitadel/internal/repository/project"
+	"github.com/zitadel/zitadel/internal/repository/user"
+
+	"github.com/zitadel/zitadel/internal/repository/milestone"
+
+	"github.com/zitadel/zitadel/internal/errors"
+	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/eventstore/handler"
+	"github.com/zitadel/zitadel/internal/eventstore/handler/crdb"
+	"github.com/zitadel/zitadel/internal/repository/instance"
+)
+
+const (
+	MilestonesProjectionTable = "projections.milestones"
+
+	MilestoneColumnInstanceID    = "instance_id"
+	MilestoneColumnMilestoneType = "milestone_type"
+	MilestoneColumnReachedAt     = "reached_at"
+	MilestoneColumnPushedAt      = "pushed_at"
+	MilestoneColumnPrimaryDomain = "primary_domain"
+)
+
+type milestoneProjection struct {
+	crdb.StatementHandler
+}
+
+func newMilestoneInstanceProjection(ctx context.Context, config crdb.StatementHandlerConfig) *milestoneProjection {
+	p := new(milestoneProjection)
+	config.ProjectionName = MilestonesProjectionTable
+	config.Reducers = p.reducers()
+	config.InitCheck = crdb.NewMultiTableCheck(
+		crdb.NewTable([]*crdb.Column{
+			crdb.NewColumn(MilestoneColumnInstanceID, crdb.ColumnTypeText),
+			crdb.NewColumn(MilestoneColumnMilestoneType, crdb.ColumnTypeEnum),
+			crdb.NewColumn(MilestoneColumnReachedAt, crdb.ColumnTypeTimestamp, crdb.Nullable()),
+			crdb.NewColumn(MilestoneColumnPushedAt, crdb.ColumnTypeTimestamp, crdb.Nullable()),
+			crdb.NewColumn(MilestoneColumnPrimaryDomain, crdb.ColumnTypeText, crdb.Nullable()),
+		},
+			crdb.NewPrimaryKey(MilestoneColumnInstanceID, MilestoneColumnMilestoneType),
+		),
+	)
+	p.StatementHandler = crdb.NewStatementHandler(ctx, config)
+	return p
+}
+
+func (p *milestoneProjection) reducers() []handler.AggregateReducer {
+	return []handler.AggregateReducer{
+		{
+			Aggregate: instance.AggregateType,
+			EventRedusers: []handler.EventReducer{
+				{
+					Event:  instance.InstanceAddedEventType,
+					Reduce: p.reduceInstanceAdded,
+				},
+				{
+					Event:  instance.InstanceDomainPrimarySetEventType,
+					Reduce: p.reduceInstanceDomainPrimarySet,
+				},
+				{
+					Event:  instance.InstanceRemovedEventType,
+					Reduce: p.reduceInstanceRemoved,
+				},
+			},
+		},
+		{
+			Aggregate: project.AggregateType,
+			EventRedusers: []handler.EventReducer{
+				{
+					Event:  project.ProjectAddedType,
+					Reduce: p.reduceProjectAdded,
+				},
+				{
+					Event:  project.ApplicationAddedType,
+					Reduce: p.reduceApplicationAdded,
+				},
+			},
+		},
+		{
+			Aggregate: user.AggregateType,
+			EventRedusers: []handler.EventReducer{
+				{
+					Event:  user.UserTokenAddedType,
+					Reduce: p.reduceUserTokenAdded,
+				},
+			},
+		},
+		{
+			Aggregate: milestone.AggregateType,
+			EventRedusers: []handler.EventReducer{
+				{
+					Event:  milestone.PushedEventType,
+					Reduce: p.milestonePushed,
+				},
+			},
+		},
+	}
+}
+
+func (p *milestoneProjection) reduceInstanceDomainPrimarySet(event eventstore.Event) (*handler.Statement, error) {
+	e, ok := event.(*instance.DomainPrimarySetEvent)
+	if !ok {
+		return nil, errors.ThrowInvalidArgumentf(nil, "HANDL-Sfrgf", "reduce.wrong.event.type %s", instance.InstanceDomainPrimarySetEventType)
+	}
+
+	var statements []func(eventstore.Event) crdb.Exec
+	for _, ms := range milestone.All() {
+		statements = append(statements, crdb.AddUpsertStatement(
+			[]handler.Column{
+				handler.NewCol(MilestoneColumnInstanceID, nil),
+				handler.NewCol(MilestoneColumnMilestoneType, nil),
+			},
+			[]handler.Column{
+				handler.NewCol(MilestoneColumnInstanceID, e.Aggregate().InstanceID),
+				handler.NewCol(MilestoneColumnMilestoneType, ms),
+				handler.NewCol(MilestoneColumnPrimaryDomain, e.Domain),
+			},
+		))
+	}
+
+	return crdb.NewMultiStatement(e, statements...), nil
+}
+
+func (p *milestoneProjection) reduceInstanceAdded(event eventstore.Event) (*handler.Statement, error) {
+	printEvent(event)
+
+	return crdb.NewNoOpStatement(event), nil
+}
+
+func (p *milestoneProjection) reduceProjectAdded(event eventstore.Event) (*handler.Statement, error) {
+	printEvent(event)
+	// ignore instance.ProjectSetEventType
+	return crdb.NewNoOpStatement(event), nil
+}
+
+func (p *milestoneProjection) reduceApplicationAdded(event eventstore.Event) (*handler.Statement, error) {
+	printEvent(event)
+	return crdb.NewNoOpStatement(event), nil
+}
+
+func (p *milestoneProjection) reduceUserTokenAdded(event eventstore.Event) (*handler.Statement, error) {
+	printEvent(event)
+	return crdb.NewNoOpStatement(event), nil
+}
+
+func (p *milestoneProjection) reduceInstanceRemoved(event eventstore.Event) (*handler.Statement, error) {
+	printEvent(event)
+	return crdb.NewNoOpStatement(event), nil
+}
+
+func (p *milestoneProjection) milestonePushed(event eventstore.Event) (*handler.Statement, error) {
+	printEvent(event)
+	return crdb.NewNoOpStatement(event), nil
+}
+
+func printEvent(event eventstore.Event) {
+	var pretty bytes.Buffer
+	if err := json.Indent(&pretty, event.DataAsBytes(), "", "    "); err != nil {
+		panic(err)
+	}
+	fmt.Println(event.Type(), pretty.String())
+}

internal/repository/milestone/events.go

@@ -1,8 +1,11 @@
+//go:
+
 package milestone
 
 import (
 	"context"
 	"encoding/json"
+	"time"
 
 	"github.com/zitadel/zitadel/internal/errors"
 	"github.com/zitadel/zitadel/internal/eventstore"
@@ -10,57 +13,16 @@ import (
 )
 
 const (
-	eventTypePrefix  = eventstore.EventType("milestone.")
-	ReachedEventType = eventTypePrefix + "reached"
-	PushedEventType  = eventTypePrefix + "pushed"
+	eventTypePrefix = eventstore.EventType("milestone.")
+	PushedEventType = eventTypePrefix + "pushed"
 )
 
-type ReachedEvent struct {
-	eventstore.BaseEvent `json:"-"`
-	MilestoneEvent       SerializableEvent `json:"milestoneEvent"`
-}
-
-func (n *ReachedEvent) Data() interface{} {
-	return n
-}
-
-func (n *ReachedEvent) UniqueConstraints() []*eventstore.EventUniqueConstraint {
-	return nil
-}
-
-func NewReachedEvent(
-	ctx context.Context,
-	newAggregateID string,
-	milestoneEvent eventstore.BaseEvent,
-) *ReachedEvent {
-	triggeringEventsAggregate := milestoneEvent.Aggregate()
-	return &ReachedEvent{
-		BaseEvent: *eventstore.NewBaseEventForPush(
-			ctx,
-			&newAggregate(newAggregateID, triggeringEventsAggregate.InstanceID, triggeringEventsAggregate.ResourceOwner).Aggregate,
-			ReachedEventType,
-		),
-		MilestoneEvent: newSerializableEvent(milestoneEvent),
-	}
-}
-
-func ReachedEventMapper(event *repository.Event) (eventstore.Event, error) {
-	e := &ReachedEvent{
-		BaseEvent: *eventstore.BaseEventFromRepo(event),
-	}
-
-	err := json.Unmarshal(event.Data, e)
-	if err != nil {
-		return nil, errors.ThrowInternal(err, "QUOTA-k56rT", "unable to unmarshal milestone reached")
-	}
-
-	return e, nil
-}
-
 type PushedEvent struct {
 	eventstore.BaseEvent `json:"-"`
-	ReachedEventSequence uint64   `json:"reachedEventSequence"`
-	Endpoints            []string `json:"endpoints"`
+	Milestone            Milestone `json:"milestone"`
+	Reached              time.Time `json:"reached"`
+	Endpoints            []string  `json:"endpoints"`
+	PrimaryDomain        string    `json:"primaryDomain"`
 }
 
 func (e *PushedEvent) Data() interface{} {
@@ -73,18 +35,22 @@ func (e *PushedEvent) UniqueConstraints() []*eventstore.EventUniqueConstraint {
 
 func NewPushedEvent(
 	ctx context.Context,
-	reachedEvent *ReachedEvent,
+	newAggregate *Aggregate,
+	milestone Milestone,
+	reached time.Time,
 	endpoints []string,
+	primaryDomain string,
 ) *PushedEvent {
-	aggregate := reachedEvent.Aggregate()
 	return &PushedEvent{
 		BaseEvent: *eventstore.NewBaseEventForPush(
 			ctx,
-			&aggregate,
+			&newAggregate.Aggregate,
 			PushedEventType,
 		),
-		ReachedEventSequence: reachedEvent.Sequence(),
-		Endpoints:            endpoints,
+		Milestone:     milestone,
+		Reached:       reached,
+		Endpoints:     endpoints,
+		PrimaryDomain: primaryDomain,
 	}
 }
 
@@ -92,11 +58,9 @@ func PushedEventMapper(event *repository.Event) (eventstore.Event, error) {
 	e := &PushedEvent{
 		BaseEvent: *eventstore.BaseEventFromRepo(event),
 	}
-
 	err := json.Unmarshal(event.Data, e)
 	if err != nil {
 		return nil, errors.ThrowInternal(err, "QUOTA-4n8vs", "unable to unmarshal milestone pushed")
 	}
-
 	return e, nil
 }

internal/repository/milestone/milestone_string.go

@@ -0,0 +1,30 @@
+// Code generated by "stringer -type=Milestone"; DO NOT EDIT.
+
+package milestone
+
+import "strconv"
+
+func _() {
+	// An "invalid array index" compiler error signifies that the constant values have changed.
+	// Re-run the stringer command to generate them again.
+	var x [1]struct{}
+	_ = x[unknown-0]
+	_ = x[InstanceCreated-1]
+	_ = x[AuthenticationSucceededOnInstance-2]
+	_ = x[ProjectCreated-3]
+	_ = x[ApplicationCreated-4]
+	_ = x[AuthenticationSucceededOnApplication-5]
+	_ = x[InstanceDeleted-6]
+	_ = x[milestonesCount-7]
+}
+
+const _Milestone_name = "unknownInstanceCreatedAuthenticationSucceededOnInstanceProjectCreatedApplicationCreatedAuthenticationSucceededOnApplicationInstanceDeletedmilestonesCount"
+
+var _Milestone_index = [...]uint8{0, 7, 22, 55, 69, 87, 123, 138, 153}
+
+func (i Milestone) String() string {
+	if i < 0 || i >= Milestone(len(_Milestone_index)-1) {
+		return "Milestone(" + strconv.FormatInt(int64(i), 10) + ")"
+	}
+	return _Milestone_name[_Milestone_index[i]:_Milestone_index[i+1]]
+}

internal/repository/milestone/milestones.go

@@ -0,0 +1,25 @@
+//go:generate stringer -type=Milestone
+
+package milestone
+
+type Milestone int
+
+const (
+	unknown Milestone = iota
+	InstanceCreated
+	AuthenticationSucceededOnInstance
+	ProjectCreated
+	ApplicationCreated
+	AuthenticationSucceededOnApplication
+	InstanceDeleted
+
+	milestonesCount
+)
+
+func All() []Milestone {
+	milestones := make([]Milestone, milestonesCount-1)
+	for i := 1; i < int(milestonesCount); i++ {
+		milestones[i] = Milestone(i)
+	}
+	return milestones
+}

internal/repository/milestone/serializable_event.go

@@ -1,15 +0,0 @@
-package milestone
-
-import "github.com/zitadel/zitadel/internal/eventstore"
-
-type SerializableEvent struct {
-	eventstore.BaseEvent `json:",inline"`
-	Data                 []byte `json:"data"`
-}
-
-func newSerializableEvent(triggeringEvent eventstore.BaseEvent) SerializableEvent {
-	return SerializableEvent{
-		BaseEvent: triggeringEvent,
-		Data:      triggeringEvent.DataAsBytes(),
-	}
-}