review fixes

internal/api/grpc/user/v2/user_integration_test.go

@@ -454,7 +454,7 @@ func TestServer_AddIDPLink(t *testing.T) {
 			args: args{
 				CTX,
 				&user.AddIDPLinkRequest{
-					UserId: Tester.Users[integration.OrgOwner].ID,
+					UserId: Tester.Users[integration.FirstInstanceUsersKey][integration.OrgOwner].ID,
 					IdpLink: &user.IDPLink{
 						IdpId:    "idpID",
 						UserId:   "userID",
@@ -470,7 +470,7 @@ func TestServer_AddIDPLink(t *testing.T) {
 			args: args{
 				CTX,
 				&user.AddIDPLinkRequest{
-					UserId: Tester.Users[integration.OrgOwner].ID,
+					UserId: Tester.Users[integration.FirstInstanceUsersKey][integration.OrgOwner].ID,
 					IdpLink: &user.IDPLink{
 						IdpId:    idpID,
 						UserId:   "userID",

internal/integration/client.go

@@ -44,8 +44,7 @@ func newClient(cc *grpc.ClientConn) Client {
 	}
 }
 
-func (t *Tester) UseIsolatedInstance(ctx context.Context) (primaryDomain, instanceID string, systemCtx, iamOwnerCtx context.Context) {
+func (t *Tester) UseIsolatedInstance(systemCtx context.Context) (primaryDomain, instanceID string, iamOwnerCtx context.Context) {
-	systemCtx = t.WithAuthorization(ctx, SystemUser)
 	primaryDomain = randString(5) + ".integration"
 	instance, err := t.Client.System.CreateInstance(systemCtx, &system.CreateInstanceRequest{
 		InstanceName: "testinstance",
@@ -61,11 +60,14 @@ func (t *Tester) UseIsolatedInstance(ctx context.Context) (primaryDomain, instan
 	if err != nil {
 		panic(err)
 	}
-	t.createClientConn(ctx, grpc.WithAuthority(primaryDomain))
+	t.createClientConn(systemCtx, grpc.WithAuthority(primaryDomain))
-	t.Users[IAMOwner] = User{
+	instanceId := instance.GetInstanceId()
-		Token: instance.GetPat(),
+	t.Users[instanceId] = map[UserType]User{
+		IAMOwner: {
+			Token: instance.GetPat(),
+		},
 	}
-	return primaryDomain, instance.GetInstanceId(), systemCtx, t.WithAuthorization(ctx, IAMOwner)
+	return primaryDomain, instanceID, t.WithInstanceAuthorization(systemCtx, IAMOwner, instanceID)
 }
 
 func (s *Tester) CreateHumanUser(ctx context.Context) *user.AddHumanUserResponse {

internal/integration/config/zitadel.yaml

@@ -39,11 +39,6 @@ Quotas:
 
 Projections:
   RequeueEvery: 1s
-  Customizations:
-    NotificationsQuotas:
-      RequeueEvery: 1s
-    Telemetry:
-      RequeueEvery: 1s
 
 DefaultInstance:
   LoginPolicy:

internal/integration/integration.go

@@ -13,6 +13,8 @@ import (
 	"sync"
 	"time"
 
+	http_util "github.com/zitadel/zitadel/internal/api/http"
+
 	"github.com/spf13/viper"
 	"github.com/zitadel/logging"
 	"github.com/zitadel/oidc/v2/pkg/client"
@@ -60,6 +62,10 @@ const (
 	SystemUser // SystemUser is a user with access to the system service.
 )
 
+const (
+	FirstInstanceUsersKey = "first"
+)
+
 // User information with a Personal Access Token.
 type User struct {
 	*query.User
@@ -72,7 +78,7 @@ type Tester struct {
 
 	Instance     authz.Instance
 	Organisation *query.Org
-	Users        map[UserType]User
+	Users        map[string]map[UserType]User
 
 	Client   Client
 	WebAuthN *webauthn.Client
@@ -133,7 +139,7 @@ const (
 	MachineUser = "integration"
 )
 
-func (s *Tester) createMachineUser(ctx context.Context) {
+func (s *Tester) createMachineUser(ctx context.Context, instanceId string) {
 	var err error
 
 	s.Instance, err = s.Queries.InstanceByHost(ctx, s.Host())
@@ -174,43 +180,43 @@ func (s *Tester) createMachineUser(ctx context.Context) {
 	_, err = s.Commands.AddPersonalAccessToken(ctx, pat)
 	logging.OnError(err).Fatal("add pat")
 
-	s.Users = map[UserType]User{
+	if s.Users == nil {
-		OrgOwner: {
+		s.Users = make(map[string]map[UserType]User)
-			User:  user,
+	}
-			Token: pat.Token,
+	if s.Users[instanceId] == nil {
-		},
+		s.Users[instanceId] = make(map[UserType]User)
+	}
+	s.Users[instanceId][OrgOwner] = User{
+		User:  user,
+		Token: pat.Token,
 	}
 }
 
 func (s *Tester) WithAuthorization(ctx context.Context, u UserType) context.Context {
+	return s.WithInstanceAuthorization(ctx, u, FirstInstanceUsersKey)
+}
+
+func (s *Tester) WithInstanceAuthorization(ctx context.Context, u UserType, instanceID string) context.Context {
 	if u == SystemUser {
 		s.ensureSystemUser()
 	}
-	return metadata.AppendToOutgoingContext(ctx, "Authorization", fmt.Sprintf("Bearer %s", s.Users[u].Token))
+	return metadata.AppendToOutgoingContext(ctx, "Authorization", fmt.Sprintf("Bearer %s", s.Users[instanceID][u].Token))
 }
 
 func (s *Tester) ensureSystemUser() {
 	const ISSUER = "tester"
-
+	if s.Users[FirstInstanceUsersKey] == nil {
-	if _, ok := s.Users[SystemUser]; ok {
+		s.Users[FirstInstanceUsersKey] = make(map[UserType]User)
+	}
+	if _, ok := s.Users[FirstInstanceUsersKey][SystemUser]; ok {
 		return
 	}
-	domain := viper.Get("ExternalDomain").(string)
+	audience := http_util.BuildOrigin(s.Host(), s.Server.Config.ExternalSecure)
-	port := viper.Get("ExternalPort").(int)
-	protocol := "http"
-	secure := viper.Get("ExternalSecure").(bool)
-	if secure {
-		protocol = "https"
-	}
-	audience := fmt.Sprintf("%s://%s:%d", protocol, domain, port)
-
 	signer, err := client.NewSignerFromPrivateKeyByte(systemUserKey, "")
 	logging.OnError(err).Fatal("system key signer")
-
 	jwt, err := client.SignedJWTProfileAssertion(ISSUER, []string{audience}, time.Hour, signer)
 	logging.OnError(err).Fatal("system key jwt")
-
+	s.Users[FirstInstanceUsersKey][SystemUser] = User{Token: jwt}
-	s.Users[SystemUser] = User{Token: jwt}
 }
 
 // Done send an interrupt signal to cleanly shutdown the server.
@@ -257,7 +263,11 @@ func NewTester(ctx context.Context) *Tester {
 	}
 	logging.OnError(err).Fatal()
 
-	tester := new(Tester)
+	tester := Tester{
+		Users: map[string]map[UserType]User{
+			FirstInstanceUsersKey: make(map[UserType]User),
+		},
+	}
 	tester.wg.Add(1)
 	go func(wg *sync.WaitGroup) {
 		logging.OnError(cmd.Execute()).Fatal()
@@ -270,10 +280,10 @@ func NewTester(ctx context.Context) *Tester {
 		logging.OnError(ctx.Err()).Fatal("waiting for integration tester server")
 	}
 	tester.createClientConn(ctx)
-	tester.createMachineUser(ctx)
+	tester.createMachineUser(ctx, FirstInstanceUsersKey)
 	tester.WebAuthN = webauthn.NewClient(tester.Config.WebAuthNName, tester.Config.ExternalDomain, "https://"+tester.Host())
 
-	return tester
+	return &tester
 }
 
 func Contexts(timeout time.Duration) (ctx, errCtx context.Context, cancel context.CancelFunc) {

internal/notification/handlers/handlers_integration_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 var (
-	CTX          context.Context
+	SystemCTX    context.Context
 	Tester       *integration.Tester
 	SystemClient system.SystemServiceClient
 	MgmtClient   management.ManagementServiceClient
@@ -23,9 +23,9 @@ var (
 func TestMain(m *testing.M) {
 	os.Exit(func() int {
 		ctx, _, cancel := integration.Contexts(5 * time.Minute)
-		CTX = ctx
 		defer cancel()
 		Tester = integration.NewTester(ctx)
+		SystemCTX = Tester.WithAuthorization(ctx, integration.SystemUser)
 		MgmtClient = Tester.Client.Mgmt
 		SystemClient = Tester.Client.System
 		defer Tester.Done()

internal/notification/handlers/telemetry_pusher.go

@@ -55,9 +55,6 @@ func NewTelemetryPusher(
 	p := new(telemetryPusher)
 	handlerCfg.ProjectionName = TelemetryProjectionTable
 	handlerCfg.Reducers = []handler.AggregateReducer{{}}
-	if telemetryCfg.Enabled {
-		handlerCfg.Reducers = p.reducers()
-	}
 	p.cfg = telemetryCfg
 	p.StatementHandler = crdb.NewStatementHandler(ctx, handlerCfg)
 	p.commands = commands

internal/notification/handlers/telemetry_pusher_integration_test.go

@@ -18,7 +18,7 @@ import (
 )
 
 func TestServer_TelemetryPushMilestones(t *testing.T) {
-	primaryDomain, instanceID, systemUserCTX, iamOwnerCtx := Tester.UseIsolatedInstance(CTX)
+	primaryDomain, instanceID, iamOwnerCtx := Tester.UseIsolatedInstance(SystemCTX)
 	bodies := make(chan []byte, 0)
 	t.Log("testing against instance with primary domain", primaryDomain)
 	mockServer := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -55,7 +55,9 @@ func TestServer_TelemetryPushMilestones(t *testing.T) {
 		t.Fatal(err)
 	}
 	awaitMilestone(t, bodies, primaryDomain, "ApplicationCreated")
-	if _, err = SystemClient.RemoveInstance(systemUserCTX, &system.RemoveInstanceRequest{InstanceId: instanceID}); err != nil {
+	// TODO: trigger and await milestone AuthenticationSucceededOnInstance
+	// TODO: trigger and await milestone AuthenticationSucceededOnApplication
+	if _, err = SystemClient.RemoveInstance(SystemCTX, &system.RemoveInstanceRequest{InstanceId: instanceID}); err != nil {
 		t.Fatal(err)
 	}
 	awaitMilestone(t, bodies, primaryDomain, "InstanceDeleted")