Merge branch 'main' into add-ns-punnycode-support

management/server/account.go

@@ -1248,7 +1248,7 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 			return nil
 		}
 
-		if err = transaction.SaveGroups(ctx, store.LockingStrengthUpdate, newGroupsToCreate); err != nil {
+		if err = transaction.SaveGroups(ctx, store.LockingStrengthUpdate, userAuth.AccountId, newGroupsToCreate); err != nil {
 			return fmt.Errorf("error saving groups: %w", err)
 		}
 
@@ -1282,7 +1282,7 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 				return fmt.Errorf("error modifying user peers in groups: %w", err)
 			}
 
-			if err = transaction.SaveGroups(ctx, store.LockingStrengthUpdate, updatedGroups); err != nil {
+			if err = transaction.SaveGroups(ctx, store.LockingStrengthUpdate, userAuth.AccountId, updatedGroups); err != nil {
 				return fmt.Errorf("error saving groups: %w", err)
 			}
 
@@ -1730,23 +1730,26 @@ func (am *DefaultAccountManager) GetStore() store.Store {
 	return am.Store
 }
 
-// Creates account by private domain.
-// Expects domain value to be a valid and a private dns domain.
-func (am *DefaultAccountManager) CreateAccountByPrivateDomain(ctx context.Context, initiatorId, domain string) (*types.Account, error) {
+func (am *DefaultAccountManager) GetOrCreateAccountByPrivateDomain(ctx context.Context, initiatorId, domain string) (*types.Account, bool, error) {
 	cancel := am.Store.AcquireGlobalLock(ctx)
 	defer cancel()
 
-	domain = strings.ToLower(domain)
-
-	count, err := am.Store.CountAccountsByPrivateDomain(ctx, domain)
-	if err != nil {
-		return nil, err
+	existingPrimaryAccountID, err := am.Store.GetAccountIDByPrivateDomain(ctx, store.LockingStrengthShare, domain)
+	if handleNotFound(err) != nil {
+		return nil, false, err
 	}
 
-	if count > 0 {
-		return nil, status.Errorf(status.InvalidArgument, "account with private domain already exists")
+	// a primary account already exists for this private domain
+	if err == nil {
+		existingAccount, err := am.Store.GetAccount(ctx, existingPrimaryAccountID)
+		if err != nil {
+			return nil, false, err
+		}
+
+		return existingAccount, false, nil
 	}
 
+	// create a new account for this private domain
 	// retry twice for new ID clashes
 	for range 2 {
 		accountId := xid.New().String()
@@ -1776,7 +1779,7 @@ func (am *DefaultAccountManager) CreateAccountByPrivateDomain(ctx context.Contex
 			Users:     users,
 			// @todo check if using the MSP owner id here is ok
 			CreatedBy:              initiatorId,
-			Domain:                 domain,
+			Domain:                 strings.ToLower(domain),
 			DomainCategory:         types.PrivateCategory,
 			IsDomainPrimaryAccount: false,
 			Routes:                 routes,
@@ -1795,19 +1798,22 @@ func (am *DefaultAccountManager) CreateAccountByPrivateDomain(ctx context.Contex
 		}
 
 		if err := newAccount.AddAllGroup(); err != nil {
-			return nil, status.Errorf(status.Internal, "failed to add all group to new account by private domain")
+			return nil, false, status.Errorf(status.Internal, "failed to add all group to new account by private domain")
 		}
 
 		if err := am.Store.SaveAccount(ctx, newAccount); err != nil {
-			log.WithContext(ctx).Errorf("failed to save new account %s by private domain: %v", newAccount.Id, err)
-			return nil, err
+			log.WithContext(ctx).WithFields(log.Fields{
+				"accountId": newAccount.Id,
+				"domain":    domain,
+			}).Errorf("failed to create new account: %v", err)
+			return nil, false, err
 		}
 
 		am.StoreEvent(ctx, initiatorId, newAccount.Id, accountId, activity.AccountCreated, nil)
-		return newAccount, nil
+		return newAccount, true, nil
 	}
 
-	return nil, status.Errorf(status.Internal, "failed to create new account by private domain")
+	return nil, false, status.Errorf(status.Internal, "failed to get or create new account by private domain")
 }
 
 func (am *DefaultAccountManager) UpdateToPrimaryAccount(ctx context.Context, accountId string) (*types.Account, error) {
@@ -1820,21 +1826,29 @@ func (am *DefaultAccountManager) UpdateToPrimaryAccount(ctx context.Context, acc
 		return account, nil
 	}
 
-	// additional check to ensure there is only one account for this domain at the time of update
-	count, err := am.Store.CountAccountsByPrivateDomain(ctx, account.Domain)
-	if err != nil {
+	existingPrimaryAccountID, err := am.Store.GetAccountIDByPrivateDomain(ctx, store.LockingStrengthShare, account.Domain)
+
+	// error is not a not found error
+	if handleNotFound(err) != nil {
 		return nil, err
 	}
 
-	if count > 1 {
-		return nil, status.Errorf(status.Internal, "more than one account exists with the same private domain")
+	// a primary account already exists for this private domain
+	if err == nil {
+		log.WithContext(ctx).WithFields(log.Fields{
+			"accountId":         accountId,
+			"existingAccountId": existingPrimaryAccountID,
+		}).Errorf("cannot update account to primary, another account already exists as primary for the same domain")
+		return nil, status.Errorf(status.Internal, "cannot update account to primary")
 	}
 
 	account.IsDomainPrimaryAccount = true
 
 	if err := am.Store.SaveAccount(ctx, account); err != nil {
-		log.WithContext(ctx).Errorf("failed to update primary account %s by private domain: %v", account.Id, err)
-		return nil, status.Errorf(status.Internal, "failed to update primary account %s by private domain", account.Id)
+		log.WithContext(ctx).WithFields(log.Fields{
+			"accountId": accountId,
+		}).Errorf("failed to update account to primary: %v", err)
+		return nil, status.Errorf(status.Internal, "failed to update account to primary")
 	}
 
 	return account, nil

management/server/account/manager.go

@@ -113,7 +113,7 @@ type Manager interface {
 	BuildUserInfosForAccount(ctx context.Context, accountID, initiatorUserID string, accountUsers []*types.User) (map[string]*types.UserInfo, error)
 	SyncUserJWTGroups(ctx context.Context, userAuth nbcontext.UserAuth) error
 	GetStore() store.Store
-	CreateAccountByPrivateDomain(ctx context.Context, initiatorId, domain string) (*types.Account, error)
+	GetOrCreateAccountByPrivateDomain(ctx context.Context, initiatorId, domain string) (*types.Account, bool, error)
 	UpdateToPrimaryAccount(ctx context.Context, accountId string) (*types.Account, error)
 	GetOwnerInfo(ctx context.Context, accountId string) (*types.UserInfo, error)
 	GetCurrentUserInfo(ctx context.Context, userAuth nbcontext.UserAuth) (*users.UserInfoWithPermissions, error)

management/server/account_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/netbirdio/netbird/management/server/activity"
 	"github.com/netbirdio/netbird/management/server/cache"
 	nbcontext "github.com/netbirdio/netbird/management/server/context"
+	"github.com/netbirdio/netbird/management/server/idp"
 	"github.com/netbirdio/netbird/management/server/integrations/port_forwarding"
 	resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types"
 	routerTypes "github.com/netbirdio/netbird/management/server/networks/routers/types"
@@ -3200,7 +3201,7 @@ func BenchmarkLoginPeer_NewPeer(b *testing.B) {
 	}
 }
 
-func Test_CreateAccountByPrivateDomain(t *testing.T) {
+func Test_GetCreateAccountByPrivateDomain(t *testing.T) {
 	manager, err := createManager(t)
 	if err != nil {
 		t.Fatal(err)
@@ -3211,9 +3212,10 @@ func Test_CreateAccountByPrivateDomain(t *testing.T) {
 	initiatorId := "test-user"
 	domain := "example.com"
 
-	account, err := manager.CreateAccountByPrivateDomain(ctx, initiatorId, domain)
+	account, created, err := manager.GetOrCreateAccountByPrivateDomain(ctx, initiatorId, domain)
 	assert.NoError(t, err)
 
+	assert.True(t, created)
 	assert.False(t, account.IsDomainPrimaryAccount)
 	assert.Equal(t, domain, account.Domain)
 	assert.Equal(t, types.PrivateCategory, account.DomainCategory)
@@ -3222,9 +3224,25 @@ func Test_CreateAccountByPrivateDomain(t *testing.T) {
 	assert.Equal(t, 0, len(account.Users))
 	assert.Equal(t, 0, len(account.SetupKeys))
 
-	// retry should fail
-	_, err = manager.CreateAccountByPrivateDomain(ctx, initiatorId, domain)
-	assert.Error(t, err)
+	// should return a new account because the previous one is not primary
+	account2, created2, err := manager.GetOrCreateAccountByPrivateDomain(ctx, initiatorId, domain)
+	assert.NoError(t, err)
+
+	assert.True(t, created2)
+	assert.False(t, account2.IsDomainPrimaryAccount)
+	assert.Equal(t, domain, account2.Domain)
+	assert.Equal(t, types.PrivateCategory, account2.DomainCategory)
+	assert.Equal(t, initiatorId, account2.CreatedBy)
+	assert.Equal(t, 1, len(account2.Groups))
+	assert.Equal(t, 0, len(account2.Users))
+	assert.Equal(t, 0, len(account2.SetupKeys))
+
+	account, err = manager.UpdateToPrimaryAccount(ctx, account.Id)
+	assert.NoError(t, err)
+	assert.True(t, account.IsDomainPrimaryAccount)
+
+	_, err = manager.UpdateToPrimaryAccount(ctx, account2.Id)
+	assert.Error(t, err, "should not be able to update a second account to primary")
 }
 
 func Test_UpdateToPrimaryAccount(t *testing.T) {
@@ -3238,14 +3256,21 @@ func Test_UpdateToPrimaryAccount(t *testing.T) {
 	initiatorId := "test-user"
 	domain := "example.com"
 
-	account, err := manager.CreateAccountByPrivateDomain(ctx, initiatorId, domain)
+	account, created, err := manager.GetOrCreateAccountByPrivateDomain(ctx, initiatorId, domain)
 	assert.NoError(t, err)
+	assert.True(t, created)
 	assert.False(t, account.IsDomainPrimaryAccount)
+	assert.Equal(t, domain, account.Domain)
 
-	// retry should fail
 	account, err = manager.UpdateToPrimaryAccount(ctx, account.Id)
 	assert.NoError(t, err)
 	assert.True(t, account.IsDomainPrimaryAccount)
+
+	account2, created2, err := manager.GetOrCreateAccountByPrivateDomain(ctx, initiatorId, domain)
+	assert.NoError(t, err)
+	assert.False(t, created2)
+	assert.True(t, account.IsDomainPrimaryAccount)
+	assert.Equal(t, account.Id, account2.Id)
 }
 
 func TestDefaultAccountManager_IsCacheCold(t *testing.T) {

management/server/activity/store/crypt.go

@@ -1,4 +1,4 @@
-package sqlite
+package store
 
 import (
 	"bytes"

management/server/activity/store/crypt_test.go

@@ -1,4 +1,4 @@
-package sqlite
+package store
 
 import (
 	"bytes"

management/server/activity/store/migration.go

@@ -1,4 +1,4 @@
-package sqlite
+package store
 
 import (
 	"context"
@@ -6,6 +6,7 @@ import (
 
 	log "github.com/sirupsen/logrus"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 
 	"github.com/netbirdio/netbird/management/server/activity"
 	"github.com/netbirdio/netbird/management/server/migration"
@@ -132,11 +133,6 @@ func migrateDuplicateDeletedUsers(ctx context.Context, db *gorm.DB) error {
 	}
 
 	if err = db.Transaction(func(tx *gorm.DB) error {
-		groupById := tx.Model(model).Select("MAX(rowid)").Group("id")
-		if err = tx.Delete(model, "rowid NOT IN (?)", groupById).Error; err != nil {
-			return err
-		}
-
 		if err = tx.Migrator().RenameTable("deleted_users", "deleted_users_old"); err != nil {
 			return err
 		}
@@ -145,12 +141,20 @@ func migrateDuplicateDeletedUsers(ctx context.Context, db *gorm.DB) error {
 			return err
 		}
 
-		if err = tx.Exec(`
-            INSERT INTO deleted_users (id, email, name, enc_algo) SELECT id, email, name, enc_algo
-              FROM deleted_users_old;`).Error; err != nil {
+		var deletedUsers []activity.DeletedUser
+		if err = tx.Table("deleted_users_old").Find(&deletedUsers).Error; err != nil {
 			return err
 		}
 
+		for _, deletedUser := range deletedUsers {
+			if err = tx.Clauses(clause.OnConflict{
+				Columns:   []clause.Column{{Name: "id"}},
+				DoUpdates: clause.AssignmentColumns([]string{"email", "name", "enc_algo"}),
+			}).Create(&deletedUser).Error; err != nil {
+				return err
+			}
+		}
+
 		return tx.Migrator().DropTable("deleted_users_old")
 	}); err != nil {
 		return err

management/server/activity/store/migration_test.go

@@ -1,17 +1,17 @@
-package sqlite
+package store
 
 import (
 	"context"
-	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"gorm.io/driver/sqlite"
+	"gorm.io/driver/postgres"
 	"gorm.io/gorm"
 
 	"github.com/netbirdio/netbird/management/server/activity"
 	"github.com/netbirdio/netbird/management/server/migration"
+	"github.com/netbirdio/netbird/management/server/testutil"
 )
 
 const (
@@ -21,8 +21,11 @@ const (
 func setupDatabase(t *testing.T) *gorm.DB {
 	t.Helper()
 
-	dbFile := filepath.Join(t.TempDir(), eventSinkDB)
-	db, err := gorm.Open(sqlite.Open(dbFile))
+	cleanup, dsn, err := testutil.CreatePostgresTestContainer()
+	require.NoError(t, err, "Failed to create Postgres test container")
+	t.Cleanup(cleanup)
+
+	db, err := gorm.Open(postgres.Open(dsn))
 	require.NoError(t, err)
 
 	sql, err := db.DB()

management/server/activity/store/sql_store.go

@@ -1,17 +1,22 @@
-package sqlite
+package store
 
 import (
 	"context"
 	"fmt"
+	"os"
 	"path/filepath"
+	"runtime"
+	"strconv"
 
 	log "github.com/sirupsen/logrus"
+	"gorm.io/driver/postgres"
 	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"
 	"gorm.io/gorm/logger"
 
 	"github.com/netbirdio/netbird/management/server/activity"
+	"github.com/netbirdio/netbird/management/server/types"
 )
 
 const (
@@ -22,6 +27,10 @@ const (
 	fallbackEmail = "unknown@unknown.com"
 
 	gcmEncAlgo = "GCM"
+
+	storeEngineEnv     = "NB_ACTIVITY_EVENT_STORE_ENGINE"
+	postgresDsnEnv     = "NB_ACTIVITY_EVENT_POSTGRES_DSN"
+	sqlMaxOpenConnsEnv = "NB_SQL_MAX_OPEN_CONNS"
 )
 
 type eventWithNames struct {
@@ -38,28 +47,19 @@ type Store struct {
 	fieldEncrypt *FieldEncrypt
 }
 
-// NewSQLiteStore creates a new Store with an event table if not exists.
-func NewSQLiteStore(ctx context.Context, dataDir string, encryptionKey string) (*Store, error) {
+// NewSqlStore creates a new Store with an event table if not exists.
+func NewSqlStore(ctx context.Context, dataDir string, encryptionKey string) (*Store, error) {
 	crypt, err := NewFieldEncrypt(encryptionKey)
 	if err != nil {
 
 		return nil, err
 	}
 
-	dbFile := filepath.Join(dataDir, eventSinkDB)
-	db, err := gorm.Open(sqlite.Open(dbFile), &gorm.Config{
-		Logger: logger.Default.LogMode(logger.Silent),
-	})
+	db, err := initDatabase(ctx, dataDir)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("initialize database: %w", err)
 	}
 
-	sql, err := db.DB()
-	if err != nil {
-		return nil, err
-	}
-	sql.SetMaxOpenConns(1)
-
 	if err = migrate(ctx, crypt, db); err != nil {
 		return nil, fmt.Errorf("events database migration: %w", err)
 	}
@@ -236,3 +236,52 @@ func (store *Store) Close(_ context.Context) error {
 	}
 	return nil
 }
+
+func initDatabase(ctx context.Context, dataDir string) (*gorm.DB, error) {
+	var dialector gorm.Dialector
+	var storeEngine = types.SqliteStoreEngine
+
+	if engine, ok := os.LookupEnv(storeEngineEnv); ok {
+		storeEngine = types.Engine(engine)
+	}
+
+	switch storeEngine {
+	case types.SqliteStoreEngine:
+		dialector = sqlite.Open(filepath.Join(dataDir, eventSinkDB))
+	case types.PostgresStoreEngine:
+		dsn, ok := os.LookupEnv(postgresDsnEnv)
+		if !ok {
+			return nil, fmt.Errorf("%s environment variable not set", postgresDsnEnv)
+		}
+		dialector = postgres.Open(dsn)
+	default:
+		return nil, fmt.Errorf("unsupported store engine: %s", storeEngine)
+	}
+	log.WithContext(ctx).Infof("using %s as activity event store engine", storeEngine)
+
+	db, err := gorm.Open(dialector, &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	if err != nil {
+		return nil, fmt.Errorf("open db connection: %w", err)
+	}
+
+	return configureConnectionPool(db, storeEngine)
+}
+
+func configureConnectionPool(db *gorm.DB, storeEngine types.Engine) (*gorm.DB, error) {
+	sqlDB, err := db.DB()
+	if err != nil {
+		return nil, err
+	}
+
+	if storeEngine == types.SqliteStoreEngine {
+		sqlDB.SetMaxOpenConns(1)
+	} else {
+		conns, err := strconv.Atoi(os.Getenv(sqlMaxOpenConnsEnv))
+		if err != nil {
+			conns = runtime.NumCPU()
+		}
+		sqlDB.SetMaxOpenConns(conns)
+	}
+
+	return db, nil
+}

management/server/activity/store/sql_store_test.go

@@ -1,4 +1,4 @@
-package sqlite
+package store
 
 import (
 	"context"
@@ -11,10 +11,10 @@ import (
 	"github.com/netbirdio/netbird/management/server/activity"
 )
 
-func TestNewSQLiteStore(t *testing.T) {
+func TestNewSqlStore(t *testing.T) {
 	dataDir := t.TempDir()
 	key, _ := GenerateKey()
-	store, err := NewSQLiteStore(context.Background(), dataDir, key)
+	store, err := NewSqlStore(context.Background(), dataDir, key)
 	if err != nil {
 		t.Fatal(err)
 		return

management/server/event.go

@@ -143,11 +143,10 @@ func (am *DefaultAccountManager) getEventsUserInfo(ctx context.Context, events [
 		return eventUserInfos, nil
 	}
 
-	return am.getEventsExternalUserInfo(ctx, externalUserIds, eventUserInfos, userId)
+	return am.getEventsExternalUserInfo(ctx, externalUserIds, eventUserInfos)
 }
 
-func (am *DefaultAccountManager) getEventsExternalUserInfo(ctx context.Context, externalUserIds []string, eventUserInfos map[string]eventUserInfo, userId string) (map[string]eventUserInfo, error) {
-	externalAccountId := ""
+func (am *DefaultAccountManager) getEventsExternalUserInfo(ctx context.Context, externalUserIds []string, eventUserInfos map[string]eventUserInfo) (map[string]eventUserInfo, error) {
 	fetched := make(map[string]struct{})
 	externalUsers := []*types.User{}
 	for _, id := range externalUserIds {
@@ -161,34 +160,30 @@ func (am *DefaultAccountManager) getEventsExternalUserInfo(ctx context.Context,
 			continue
 		}
 
-		if externalAccountId != "" && externalAccountId != externalUser.AccountID {
-			return nil, fmt.Errorf("multiple external user accounts in events")
-		}
-
-		if externalAccountId == "" {
-			externalAccountId = externalUser.AccountID
-		}
-
 		fetched[id] = struct{}{}
 		externalUsers = append(externalUsers, externalUser)
 	}
 
-	// if we couldn't determine an account, return what we have
-	if externalAccountId == "" {
-		log.WithContext(ctx).Warnf("failed to determine external user account from users: %v", externalUserIds)
-		return eventUserInfos, nil
+	usersByExternalAccount := map[string][]*types.User{}
+	for _, u := range externalUsers {
+		if _, ok := usersByExternalAccount[u.AccountID]; !ok {
+			usersByExternalAccount[u.AccountID] = make([]*types.User, 0)
+		}
+		usersByExternalAccount[u.AccountID] = append(usersByExternalAccount[u.AccountID], u)
 	}
 
-	externalUserInfos, err := am.BuildUserInfosForAccount(ctx, externalAccountId, userId, externalUsers)
-	if err != nil {
-		return nil, err
-	}
+	for externalAccountId, externalUsers := range usersByExternalAccount {
+		externalUserInfos, err := am.BuildUserInfosForAccount(ctx, externalAccountId, "", externalUsers)
+		if err != nil {
+			return nil, err
+		}
 
-	for i, k := range externalUserInfos {
-		eventUserInfos[i] = eventUserInfo{
-			email:     k.Email,
-			name:      k.Name,
-			accountId: externalAccountId,
+		for i, k := range externalUserInfos {
+			eventUserInfos[i] = eventUserInfo{
+				email:     k.Email,
+				name:      k.Name,
+				accountId: externalAccountId,
+			}
 		}
 	}
 

management/server/group.go

@@ -116,7 +116,7 @@ func (am *DefaultAccountManager) SaveGroups(ctx context.Context, accountID, user
 			return err
 		}
 
-		return transaction.SaveGroups(ctx, store.LockingStrengthUpdate, groupsToSave)
+		return transaction.SaveGroups(ctx, store.LockingStrengthUpdate, accountID, groupsToSave)
 	})
 	if err != nil {
 		return err

management/server/http/api/openapi.yml

@@ -1925,13 +1925,71 @@ components:
         - os
         - address
         - dns_label
-    NetworkTrafficEvent:
+    NetworkTrafficUser:
       type: object
       properties:
         id:
           type: string
-          description: "ID of the event. Unique."
-          example: "18e204d6-f7c6-405d-8025-70becb216add"
+          description: "UserID is the ID of the user that initiated the event (can be empty as not every event is user-initiated)."
+          example: "google-oauth2|123456789012345678901"
+        email:
+          type: string
+          description: "Email of the user who initiated the event (if any)."
+          example: "alice@netbird.io"
+        name:
+            type: string
+            description: "Name of the user who initiated the event (if any)."
+            example: "Alice Smith"
+      required:
+        - id
+        - email
+        - name
+    NetworkTrafficPolicy:
+      type: object
+      properties:
+        id:
+          type: string
+          description: "ID of the policy that allowed this event."
+          example: "ch8i4ug6lnn4g9hqv7m0"
+        name:
+          type: string
+          description: "Name of the policy that allowed this event."
+          example: "All to All"
+      required:
+        - id
+        - name
+    NetworkTrafficICMP:
+      type: object
+      properties:
+        type:
+          type: integer
+          description: "ICMP type (if applicable)."
+          example: 8
+        code:
+          type: integer
+          description: "ICMP code (if applicable)."
+          example: 0
+      required:
+        - type
+        - code
+    NetworkTrafficSubEvent:
+      type: object
+      properties:
+        type:
+          type: string
+          description: Type of the event (e.g., TYPE_UNKNOWN, TYPE_START, TYPE_END, TYPE_DROP).
+          example: TYPE_START
+        timestamp:
+          type: string
+          format: date-time
+          description: Timestamp of the event as sent by the peer.
+          example: 2025-03-20T16:23:58.125397Z
+      required:
+        - type
+        - timestamp
+    NetworkTrafficEvent:
+      type: object
+      properties:
         flow_id:
           type: string
           description: "FlowID is the ID of the connection flow. Not unique because it can be the same for multiple events (e.g., start and end of the connection)."
@@ -1940,43 +1998,20 @@ components:
           type: string
           description: "ID of the reporter of the event (e.g., the peer that reported the event)."
           example: "ch8i4ug6lnn4g9hqv7m0"
-        timestamp:
-          type: string
-          format: date-time
-          description: "Timestamp of the event. Send by the peer."
-          example: "2025-03-20T16:23:58.125397Z"
-        receive_timestamp:
-          type: string
-          format: date-time
-          description: "Timestamp when the event was received by our API."
-          example: "2025-03-20T16:23:58.125397Z"
         source:
           $ref: '#/components/schemas/NetworkTrafficEndpoint'
-        user_id:
-          type: string
-          nullable: true
-          description: "UserID is the ID of the user that initiated the event (can be empty as not every event is user-initiated)."
-          example: "google-oauth2|123456789012345678901"
-        user_email:
-          type: string
-          nullable: true
-          description: "Email of the user who initiated the event (if any)."
-          example: "alice@netbird.io"
-        user_name:
-          type: string
-          nullable: true
-          description: "Name of the user who initiated the event (if any)."
-          example: "Alice Smith"
         destination:
           $ref: '#/components/schemas/NetworkTrafficEndpoint'
+        user:
+          $ref: '#/components/schemas/NetworkTrafficUser'
+        policy:
+          $ref: '#/components/schemas/NetworkTrafficPolicy'
+        icmp:
+          $ref: '#/components/schemas/NetworkTrafficICMP'
         protocol:
           type: integer
           description: "Protocol is the protocol of the traffic (e.g. 1 = ICMP, 6 = TCP, 17 = UDP, etc.)."
           example: 6
-        type:
-          type: string
-          description: "Type of the event (e.g. TYPE_UNKNOWN, TYPE_START, TYPE_END, TYPE_DROP)."
-          example: "TYPE_START"
         direction:
           type: string
           description: "Direction of the traffic (e.g. DIRECTION_UNKNOWN, INGRESS, EGRESS)."
@@ -1997,43 +2032,28 @@ components:
           type: integer
           description: "Number of packets transmitted."
           example: 5
-        policy_id:
-          type: string
-          description: "ID of the policy that allowed this event."
-          example: "ch8i4ug6lnn4g9hqv7m0"
-        policy_name:
-          type: string
-          description: "Name of the policy that allowed this event."
-          example: "All to All"
-        icmp_type:
-          type: integer
-          description: "ICMP type (if applicable)."
-          example: 8
-        icmp_code:
-          type: integer
-          description: "ICMP code (if applicable)."
-          example: 0
+        events:
+          type: array
+          description: "List of events that are correlated to this flow (e.g., start, end)."
+          items:
+            $ref: '#/components/schemas/NetworkTrafficSubEvent'
       required:
         - id
         - flow_id
         - reporter_id
-        - timestamp
         - receive_timestamp
         - source
-        - user_id
-        - user_email
         - destination
+        - user
+        - policy
+        - icmp
         - protocol
-        - type
         - direction
         - rx_bytes
         - rx_packets
         - tx_bytes
         - tx_packets
-        - policy_id
-        - policy_name
-        - icmp_type
-        - icmp_code
+        - events
     NetworkTrafficEventsResponse:
       type: object
       properties:
@@ -4048,6 +4068,31 @@ paths:
           "$ref": "#/components/responses/forbidden"
         '500':
           "$ref": "#/components/responses/internal_error"
+  /api/networks/routers:
+    get:
+      summary: List all Network Routers
+      description: Returns a list of all routers in a network
+      tags: [ Networks ]
+      security:
+        - BearerAuth: [ ]
+        - TokenAuth: [ ]
+      responses:
+        '200':
+          description: A JSON Array of Routers
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/NetworkRouter'
+        '400':
+          "$ref": "#/components/responses/bad_request"
+        '401':
+          "$ref": "#/components/responses/requires_authentication"
+        '403':
+          "$ref": "#/components/responses/forbidden"
+        '500':
+          "$ref": "#/components/responses/internal_error"
   /api/dns/nameservers:
     get:
       summary: List all Nameserver Groups

management/server/http/api/types.gen.go

@@ -883,30 +883,17 @@ type NetworkTrafficEvent struct {
 	// Direction Direction of the traffic (e.g. DIRECTION_UNKNOWN, INGRESS, EGRESS).
 	Direction string `json:"direction"`
 
+	// Events List of events that are correlated to this flow (e.g., start, end).
+	Events []NetworkTrafficSubEvent `json:"events"`
+
 	// FlowId FlowID is the ID of the connection flow. Not unique because it can be the same for multiple events (e.g., start and end of the connection).
-	FlowId string `json:"flow_id"`
-
-	// IcmpCode ICMP code (if applicable).
-	IcmpCode int `json:"icmp_code"`
-
-	// IcmpType ICMP type (if applicable).
-	IcmpType int `json:"icmp_type"`
-
-	// Id ID of the event. Unique.
-	Id string `json:"id"`
-
-	// PolicyId ID of the policy that allowed this event.
-	PolicyId string `json:"policy_id"`
-
-	// PolicyName Name of the policy that allowed this event.
-	PolicyName string `json:"policy_name"`
+	FlowId string               `json:"flow_id"`
+	Icmp   NetworkTrafficICMP   `json:"icmp"`
+	Policy NetworkTrafficPolicy `json:"policy"`
 
 	// Protocol Protocol is the protocol of the traffic (e.g. 1 = ICMP, 6 = TCP, 17 = UDP, etc.).
 	Protocol int `json:"protocol"`
 
-	// ReceiveTimestamp Timestamp when the event was received by our API.
-	ReceiveTimestamp time.Time `json:"receive_timestamp"`
-
 	// ReporterId ID of the reporter of the event (e.g., the peer that reported the event).
 	ReporterId string `json:"reporter_id"`
 
@@ -917,26 +904,12 @@ type NetworkTrafficEvent struct {
 	RxPackets int                    `json:"rx_packets"`
 	Source    NetworkTrafficEndpoint `json:"source"`
 
-	// Timestamp Timestamp of the event. Send by the peer.
-	Timestamp time.Time `json:"timestamp"`
-
 	// TxBytes Number of bytes transmitted.
 	TxBytes int `json:"tx_bytes"`
 
 	// TxPackets Number of packets transmitted.
-	TxPackets int `json:"tx_packets"`
-
-	// Type Type of the event (e.g. TYPE_UNKNOWN, TYPE_START, TYPE_END, TYPE_DROP).
-	Type string `json:"type"`
-
-	// UserEmail Email of the user who initiated the event (if any).
-	UserEmail *string `json:"user_email"`
-
-	// UserId UserID is the ID of the user that initiated the event (can be empty as not every event is user-initiated).
-	UserId *string `json:"user_id"`
-
-	// UserName Name of the user who initiated the event (if any).
-	UserName *string `json:"user_name"`
+	TxPackets int                `json:"tx_packets"`
+	User      NetworkTrafficUser `json:"user"`
 }
 
 // NetworkTrafficEventsResponse defines model for NetworkTrafficEventsResponse.
@@ -957,6 +930,15 @@ type NetworkTrafficEventsResponse struct {
 	TotalRecords int `json:"total_records"`
 }
 
+// NetworkTrafficICMP defines model for NetworkTrafficICMP.
+type NetworkTrafficICMP struct {
+	// Code ICMP code (if applicable).
+	Code int `json:"code"`
+
+	// Type ICMP type (if applicable).
+	Type int `json:"type"`
+}
+
 // NetworkTrafficLocation defines model for NetworkTrafficLocation.
 type NetworkTrafficLocation struct {
 	// CityName Name of the city (if known).
@@ -966,6 +948,36 @@ type NetworkTrafficLocation struct {
 	CountryCode string `json:"country_code"`
 }
 
+// NetworkTrafficPolicy defines model for NetworkTrafficPolicy.
+type NetworkTrafficPolicy struct {
+	// Id ID of the policy that allowed this event.
+	Id string `json:"id"`
+
+	// Name Name of the policy that allowed this event.
+	Name string `json:"name"`
+}
+
+// NetworkTrafficSubEvent defines model for NetworkTrafficSubEvent.
+type NetworkTrafficSubEvent struct {
+	// Timestamp Timestamp of the event as sent by the peer.
+	Timestamp time.Time `json:"timestamp"`
+
+	// Type Type of the event (e.g., TYPE_UNKNOWN, TYPE_START, TYPE_END, TYPE_DROP).
+	Type string `json:"type"`
+}
+
+// NetworkTrafficUser defines model for NetworkTrafficUser.
+type NetworkTrafficUser struct {
+	// Email Email of the user who initiated the event (if any).
+	Email string `json:"email"`
+
+	// Id UserID is the ID of the user that initiated the event (can be empty as not every event is user-initiated).
+	Id string `json:"id"`
+
+	// Name Name of the user who initiated the event (if any).
+	Name string `json:"name"`
+}
+
 // OSVersionCheck Posture check for the version of operating system
 type OSVersionCheck struct {
 	// Android Posture check for the version of operating system

management/server/http/handlers/networks/routers_handler.go

@@ -19,7 +19,8 @@ type routersHandler struct {
 
 func addRouterEndpoints(routersManager routers.Manager, router *mux.Router) {
 	routersHandler := newRoutersHandler(routersManager)
-	router.HandleFunc("/networks/{networkId}/routers", routersHandler.getAllRouters).Methods("GET", "OPTIONS")
+	router.HandleFunc("/networks/routers", routersHandler.getAllRouters).Methods("GET", "OPTIONS")
+	router.HandleFunc("/networks/{networkId}/routers", routersHandler.getNetworkRouters).Methods("GET", "OPTIONS")
 	router.HandleFunc("/networks/{networkId}/routers", routersHandler.createRouter).Methods("POST", "OPTIONS")
 	router.HandleFunc("/networks/{networkId}/routers/{routerId}", routersHandler.getRouter).Methods("GET", "OPTIONS")
 	router.HandleFunc("/networks/{networkId}/routers/{routerId}", routersHandler.updateRouter).Methods("PUT", "OPTIONS")
@@ -41,6 +42,31 @@ func (h *routersHandler) getAllRouters(w http.ResponseWriter, r *http.Request) {
 
 	accountID, userID := userAuth.AccountId, userAuth.UserId
 
+	routersMap, err := h.routersManager.GetAllRoutersInAccount(r.Context(), accountID, userID)
+	if err != nil {
+		util.WriteError(r.Context(), err, w)
+		return
+	}
+
+	routersResponse := make([]*api.NetworkRouter, 0)
+	for _, routers := range routersMap {
+		for _, router := range routers {
+			routersResponse = append(routersResponse, router.ToAPIResponse())
+		}
+	}
+
+	util.WriteJSONObject(r.Context(), w, routersResponse)
+}
+
+func (h *routersHandler) getNetworkRouters(w http.ResponseWriter, r *http.Request) {
+	userAuth, err := nbcontext.GetUserAuthFromContext(r.Context())
+	if err != nil {
+		util.WriteError(r.Context(), err, w)
+		return
+	}
+
+	accountID, userID := userAuth.AccountId, userAuth.UserId
+
 	networkID := mux.Vars(r)["networkId"]
 	routers, err := h.routersManager.GetAllRoutersInNetwork(r.Context(), accountID, userID, networkID)
 	if err != nil {

management/server/mock_server/account_mock.go

@@ -113,11 +113,12 @@ type MockAccountManager struct {
 	DeleteSetupKeyFunc                  func(ctx context.Context, accountID, userID, keyID string) error
 	BuildUserInfosForAccountFunc        func(ctx context.Context, accountID, initiatorUserID string, accountUsers []*types.User) (map[string]*types.UserInfo, error)
 	GetStoreFunc                        func() store.Store
-	CreateAccountByPrivateDomainFunc    func(ctx context.Context, initiatorId, domain string) (*types.Account, error)
 	UpdateToPrimaryAccountFunc          func(ctx context.Context, accountId string) (*types.Account, error)
 	GetOwnerInfoFunc                    func(ctx context.Context, accountID string) (*types.UserInfo, error)
 	GetCurrentUserInfoFunc              func(ctx context.Context, userAuth nbcontext.UserAuth) (*users.UserInfoWithPermissions, error)
 	GetAccountMetaFunc                  func(ctx context.Context, accountID, userID string) (*types.AccountMeta, error)
+
+	GetOrCreateAccountByPrivateDomainFunc func(ctx context.Context, initiatorId, domain string) (*types.Account, bool, error)
 }
 
 func (am *MockAccountManager) UpdateAccountPeers(ctx context.Context, accountID string) {
@@ -862,11 +863,11 @@ func (am *MockAccountManager) GetStore() store.Store {
 	return nil
 }
 
-func (am *MockAccountManager) CreateAccountByPrivateDomain(ctx context.Context, initiatorId, domain string) (*types.Account, error) {
-	if am.CreateAccountByPrivateDomainFunc != nil {
-		return am.CreateAccountByPrivateDomainFunc(ctx, initiatorId, domain)
+func (am *MockAccountManager) GetOrCreateAccountByPrivateDomain(ctx context.Context, initiatorId, domain string) (*types.Account, bool, error) {
+	if am.GetOrCreateAccountByPrivateDomainFunc != nil {
+		return am.GetOrCreateAccountByPrivateDomainFunc(ctx, initiatorId, domain)
 	}
-	return nil, status.Errorf(codes.Unimplemented, "method CreateAccountByPrivateDomain is not implemented")
+	return nil, false, status.Errorf(codes.Unimplemented, "method GetOrCreateAccountByPrivateDomainFunc is not implemented")
 }
 
 func (am *MockAccountManager) UpdateToPrimaryAccount(ctx context.Context, accountId string) (*types.Account, error) {

management/server/peer.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/netbirdio/netbird/management/domain"
 	"github.com/netbirdio/netbird/management/server/geolocation"
+	routerTypes "github.com/netbirdio/netbird/management/server/networks/routers/types"
 	"github.com/netbirdio/netbird/management/server/permissions/modules"
 	"github.com/netbirdio/netbird/management/server/permissions/operations"
 
@@ -352,7 +353,7 @@ func (am *DefaultAccountManager) DeletePeer(ctx context.Context, accountID, peer
 			return err
 		}
 
-		if err = am.validatePeerDelete(ctx, accountID, peerID); err != nil {
+		if err = am.validatePeerDelete(ctx, transaction, accountID, peerID); err != nil {
 			return err
 		}
 
@@ -1543,7 +1544,7 @@ func ConvertSliceToMap(existingLabels []string) map[string]struct{} {
 }
 
 // validatePeerDelete checks if the peer can be deleted.
-func (am *DefaultAccountManager) validatePeerDelete(ctx context.Context, accountId, peerId string) error {
+func (am *DefaultAccountManager) validatePeerDelete(ctx context.Context, transaction store.Store, accountId, peerId string) error {
 	linkedInIngressPorts, err := am.proxyController.IsPeerInIngressPorts(ctx, accountId, peerId)
 	if err != nil {
 		return err
@@ -1553,5 +1554,27 @@ func (am *DefaultAccountManager) validatePeerDelete(ctx context.Context, account
 		return status.Errorf(status.PreconditionFailed, "peer is linked to ingress ports: %s", peerId)
 	}
 
+	linked, router := isPeerLinkedToNetworkRouter(ctx, transaction, accountId, peerId)
+	if linked {
+		return status.Errorf(status.PreconditionFailed, "peer is linked to a network router: %s", router.ID)
+	}
+
 	return nil
 }
+
+// isPeerLinkedToNetworkRouter checks if a peer is linked to any network router in the account.
+func isPeerLinkedToNetworkRouter(ctx context.Context, transaction store.Store, accountID string, peerID string) (bool, *routerTypes.NetworkRouter) {
+	routers, err := transaction.GetNetworkRoutersByAccountID(ctx, store.LockingStrengthShare, accountID)
+	if err != nil {
+		log.WithContext(ctx).Errorf("error retrieving network routers while checking peer linkage: %v", err)
+		return false, nil
+	}
+
+	for _, router := range routers {
+		if router.Peer == peerID {
+			return true, router
+		}
+	}
+
+	return false, nil
+}

management/server/posture_checks_test.go

@@ -455,7 +455,7 @@ func TestArePostureCheckChangesAffectPeers(t *testing.T) {
 		AccountID: account.Id,
 		Peers:     []string{},
 	}
-	err = manager.Store.SaveGroups(context.Background(), store.LockingStrengthUpdate, []*types.Group{groupA, groupB})
+	err = manager.Store.SaveGroups(context.Background(), store.LockingStrengthUpdate, account.Id, []*types.Group{groupA, groupB})
 	require.NoError(t, err, "failed to save groups")
 
 	postureCheckA := &posture.Checks{

management/server/store/sql_store.go

@@ -448,12 +448,20 @@ func (s *SqlStore) SaveUser(ctx context.Context, lockStrength LockingStrength, u
 }
 
 // SaveGroups saves the given list of groups to the database.
-func (s *SqlStore) SaveGroups(ctx context.Context, lockStrength LockingStrength, groups []*types.Group) error {
+func (s *SqlStore) SaveGroups(ctx context.Context, lockStrength LockingStrength, accountID string, groups []*types.Group) error {
 	if len(groups) == 0 {
 		return nil
 	}
 
-	result := s.db.Clauses(clause.Locking{Strength: string(lockStrength)}, clause.OnConflict{UpdateAll: true}).Create(&groups)
+	result := s.db.
+		Clauses(
+			clause.Locking{Strength: string(lockStrength)},
+			clause.OnConflict{
+				Where:     clause.Where{Exprs: []clause.Expression{clause.Eq{Column: "groups.account_id", Value: accountID}}},
+				UpdateAll: true,
+			},
+		).
+		Create(&groups)
 	if result.Error != nil {
 		return status.Errorf(status.Internal, "failed to save groups to store: %v", result.Error)
 	}

management/server/store/sql_store_test.go

@@ -1324,11 +1324,11 @@ func TestSqlStore_SaveGroups(t *testing.T) {
 			Peers:     []string{"peer3", "peer4"},
 		},
 	}
-	err = store.SaveGroups(context.Background(), LockingStrengthUpdate, groups)
+	err = store.SaveGroups(context.Background(), LockingStrengthUpdate, accountID, groups)
 	require.NoError(t, err)
 
 	groups[1].Peers = []string{}
-	err = store.SaveGroups(context.Background(), LockingStrengthUpdate, groups)
+	err = store.SaveGroups(context.Background(), LockingStrengthUpdate, accountID, groups)
 	require.NoError(t, err)
 
 	group, err := store.GetGroupByID(context.Background(), LockingStrengthShare, accountID, groups[1].ID)
@@ -3240,7 +3240,7 @@ func TestSqlStore_SaveGroups_LargeBatch(t *testing.T) {
 		})
 	}
 
-	err = store.SaveGroups(context.Background(), LockingStrengthUpdate, groupsToSave)
+	err = store.SaveGroups(context.Background(), LockingStrengthUpdate, accountID, groupsToSave)
 	require.NoError(t, err)
 
 	accountGroups, err = store.GetAccountGroups(context.Background(), LockingStrengthShare, accountID)

management/server/store/store.go

@@ -98,7 +98,7 @@ type Store interface {
 	GetGroupByID(ctx context.Context, lockStrength LockingStrength, accountID, groupID string) (*types.Group, error)
 	GetGroupByName(ctx context.Context, lockStrength LockingStrength, groupName, accountID string) (*types.Group, error)
 	GetGroupsByIDs(ctx context.Context, lockStrength LockingStrength, accountID string, groupIDs []string) (map[string]*types.Group, error)
-	SaveGroups(ctx context.Context, lockStrength LockingStrength, groups []*types.Group) error
+	SaveGroups(ctx context.Context, lockStrength LockingStrength, accountID string, groups []*types.Group) error
 	SaveGroup(ctx context.Context, lockStrength LockingStrength, group *types.Group) error
 	DeleteGroup(ctx context.Context, lockStrength LockingStrength, accountID, groupID string) error
 	DeleteGroups(ctx context.Context, strength LockingStrength, accountID string, groupIDs []string) error
@@ -365,11 +365,14 @@ func NewTestStoreFromSQL(ctx context.Context, filename string, dataDir string) (
 		return nil, nil, fmt.Errorf("failed to add all group to account: %v", err)
 	}
 
+	var sqlStore Store
+	var cleanup func()
+
 	maxRetries := 2
 	for i := 0; i < maxRetries; i++ {
-		sqlStore, cleanUp, err := getSqlStoreEngine(ctx, store, kind)
+		sqlStore, cleanup, err = getSqlStoreEngine(ctx, store, kind)
 		if err == nil {
-			return sqlStore, cleanUp, nil
+			return sqlStore, cleanup, nil
 		}
 		if i < maxRetries-1 {
 			time.Sleep(100 * time.Millisecond)
@@ -427,16 +430,16 @@ func getSqlStoreEngine(ctx context.Context, store *SqlStore, kind types.Engine)
 }
 
 func newReusedPostgresStore(ctx context.Context, store *SqlStore, kind types.Engine) (*SqlStore, func(), error) {
-	if envDsn, ok := os.LookupEnv(postgresDsnEnv); !ok || envDsn == "" {
+	dsn, ok := os.LookupEnv(postgresDsnEnv)
+	if !ok || dsn == "" {
 		var err error
-		_, err = testutil.CreatePostgresTestContainer()
+		_, dsn, err = testutil.CreatePostgresTestContainer()
 		if err != nil {
 			return nil, nil, err
 		}
 	}
 
-	dsn, ok := os.LookupEnv(postgresDsnEnv)
-	if !ok {
+	if dsn == "" {
 		return nil, nil, fmt.Errorf("%s is not set", postgresDsnEnv)
 	}
 
@@ -447,28 +450,28 @@ func newReusedPostgresStore(ctx context.Context, store *SqlStore, kind types.Eng
 
 	dsn, cleanup, err := createRandomDB(dsn, db, kind)
 	if err != nil {
-		return nil, cleanup, err
+		return nil, nil, err
 	}
 
 	store, err = NewPostgresqlStoreFromSqlStore(ctx, store, dsn, nil)
 	if err != nil {
-		return nil, cleanup, err
+		return nil, nil, err
 	}
 
 	return store, cleanup, nil
 }
 
 func newReusedMysqlStore(ctx context.Context, store *SqlStore, kind types.Engine) (*SqlStore, func(), error) {
-	if envDsn, ok := os.LookupEnv(mysqlDsnEnv); !ok || envDsn == "" {
+	dsn, ok := os.LookupEnv(mysqlDsnEnv)
+	if !ok || dsn == "" {
 		var err error
-		_, err = testutil.CreateMysqlTestContainer()
+		_, dsn, err = testutil.CreateMysqlTestContainer()
 		if err != nil {
 			return nil, nil, err
 		}
 	}
 
-	dsn, ok := os.LookupEnv(mysqlDsnEnv)
-	if !ok {
+	if dsn == "" {
 		return nil, nil, fmt.Errorf("%s is not set", mysqlDsnEnv)
 	}
 
@@ -479,7 +482,7 @@ func newReusedMysqlStore(ctx context.Context, store *SqlStore, kind types.Engine
 
 	dsn, cleanup, err := createRandomDB(dsn, db, kind)
 	if err != nil {
-		return nil, cleanup, err
+		return nil, nil, err
 	}
 
 	store, err = NewMysqlStoreFromSqlStore(ctx, store, dsn, nil)

management/server/testutil/store.go

@@ -5,7 +5,6 @@ package testutil
 
 import (
 	"context"
-	"os"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -16,11 +15,25 @@ import (
 	"github.com/testcontainers/testcontainers-go/wait"
 )
 
+var (
+	pgContainer    *postgres.PostgresContainer
+	mysqlContainer *mysql.MySQLContainer
+)
+
 // CreateMysqlTestContainer creates a new MySQL container for testing.
-func CreateMysqlTestContainer() (func(), error) {
+func CreateMysqlTestContainer() (func(), string, error) {
 	ctx := context.Background()
 
-	myContainer, err := mysql.RunContainer(ctx,
+	if mysqlContainer != nil {
+		connStr, err := mysqlContainer.ConnectionString(ctx)
+		if err != nil {
+			return nil, "", err
+		}
+		return noOpCleanup, connStr, nil
+	}
+
+	var err error
+	mysqlContainer, err = mysql.RunContainer(ctx,
 		testcontainers.WithImage("mlsmaycon/warmed-mysql:8"),
 		mysql.WithDatabase("testing"),
 		mysql.WithUsername("root"),
@@ -31,31 +44,42 @@ func CreateMysqlTestContainer() (func(), error) {
 		),
 	)
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 
 	cleanup := func() {
-		os.Unsetenv("NETBIRD_STORE_ENGINE_MYSQL_DSN")
-		timeoutCtx, cancelFunc := context.WithTimeout(ctx, 1*time.Second)
-		defer cancelFunc()
-		if err = myContainer.Terminate(timeoutCtx); err != nil {
-			log.WithContext(ctx).Warnf("failed to stop mysql container %s: %s", myContainer.GetContainerID(), err)
+		if mysqlContainer != nil {
+			timeoutCtx, cancelFunc := context.WithTimeout(ctx, 1*time.Second)
+			defer cancelFunc()
+			if err = mysqlContainer.Terminate(timeoutCtx); err != nil {
+				log.WithContext(ctx).Warnf("failed to stop mysql container %s: %s", mysqlContainer.GetContainerID(), err)
+			}
+			mysqlContainer = nil // reset the container to allow recreation
 		}
 	}
 
-	talksConn, err := myContainer.ConnectionString(ctx)
+	talksConn, err := mysqlContainer.ConnectionString(ctx)
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 
-	return cleanup, os.Setenv("NETBIRD_STORE_ENGINE_MYSQL_DSN", talksConn)
+	return cleanup, talksConn, nil
 }
 
 // CreatePostgresTestContainer creates a new PostgreSQL container for testing.
-func CreatePostgresTestContainer() (func(), error) {
+func CreatePostgresTestContainer() (func(), string, error) {
 	ctx := context.Background()
 
-	pgContainer, err := postgres.RunContainer(ctx,
+	if pgContainer != nil {
+		connStr, err := pgContainer.ConnectionString(ctx)
+		if err != nil {
+			return nil, "", err
+		}
+		return noOpCleanup, connStr, nil
+	}
+
+	var err error
+	pgContainer, err = postgres.RunContainer(ctx,
 		testcontainers.WithImage("postgres:16-alpine"),
 		postgres.WithDatabase("netbird"),
 		postgres.WithUsername("root"),
@@ -66,24 +90,31 @@ func CreatePostgresTestContainer() (func(), error) {
 		),
 	)
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 
 	cleanup := func() {
-		os.Unsetenv("NETBIRD_STORE_ENGINE_POSTGRES_DSN")
-		timeoutCtx, cancelFunc := context.WithTimeout(ctx, 1*time.Second)
-		defer cancelFunc()
-		if err = pgContainer.Terminate(timeoutCtx); err != nil {
-			log.WithContext(ctx).Warnf("failed to stop postgres container %s: %s", pgContainer.GetContainerID(), err)
+		if pgContainer != nil {
+			timeoutCtx, cancelFunc := context.WithTimeout(ctx, 1*time.Second)
+			defer cancelFunc()
+			if err = pgContainer.Terminate(timeoutCtx); err != nil {
+				log.WithContext(ctx).Warnf("failed to stop postgres container %s: %s", pgContainer.GetContainerID(), err)
+			}
+			pgContainer = nil // reset the container to allow recreation
 		}
+
 	}
 
 	talksConn, err := pgContainer.ConnectionString(ctx)
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 
-	return cleanup, os.Setenv("NETBIRD_STORE_ENGINE_POSTGRES_DSN", talksConn)
+	return cleanup, talksConn, nil
+}
+
+func noOpCleanup() {
+	// no-op
 }
 
 // CreateRedisTestContainer creates a new Redis container for testing.

management/server/testutil/store_ios.go

@@ -3,16 +3,16 @@
 
 package testutil
 
-func CreatePostgresTestContainer() (func(), error) {
+func CreatePostgresTestContainer() (func(), string, error) {
 	return func() {
 		// Empty function for Postgres
-	}, nil
+	}, "", nil
 }
 
-func CreateMysqlTestContainer() (func(), error) {
+func CreateMysqlTestContainer() (func(), string, error) {
 	return func() {
 		// Empty function for MySQL
-	}, nil
+	}, "", nil
 }
 
 func CreateRedisTestContainer() (func(), string, error) {

management/server/types/settings.go

@@ -45,7 +45,7 @@ type Settings struct {
 	// Extra is a dictionary of Account settings
 	Extra *ExtraSettings `gorm:"embedded;embeddedPrefix:extra_"`
 
-	// LazyConnectionEnabled indicates wether the experimental feature is enabled or disabled
+	// LazyConnectionEnabled indicates if the experimental feature is enabled or disabled
 	LazyConnectionEnabled bool `gorm:"default:false"`
 }
 

management/server/user.go

@@ -531,9 +531,13 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID,
 		groupsMap[group.ID] = group
 	}
 
-	initiatorUser, err := am.Store.GetUserByUserID(ctx, store.LockingStrengthShare, initiatorUserID)
-	if err != nil {
-		return nil, err
+	var initiatorUser *types.User
+	if initiatorUserID != activity.SystemInitiator {
+		result, err := am.Store.GetUserByUserID(ctx, store.LockingStrengthShare, initiatorUserID)
+		if err != nil {
+			return nil, err
+		}
+		initiatorUser = result
 	}
 
 	err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
@@ -543,7 +547,7 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID,
 			}
 
 			userHadPeers, updatedUser, userPeersToExpire, userEvents, err := am.processUserUpdate(
-				ctx, transaction, groupsMap, accountID, initiatorUser, update, addIfNotExists, settings,
+				ctx, transaction, groupsMap, accountID, initiatorUserID, initiatorUser, update, addIfNotExists, settings,
 			)
 			if err != nil {
 				return fmt.Errorf("failed to process user update: %w", err)
@@ -629,7 +633,7 @@ func (am *DefaultAccountManager) prepareUserUpdateEvents(ctx context.Context, ac
 }
 
 func (am *DefaultAccountManager) processUserUpdate(ctx context.Context, transaction store.Store, groupsMap map[string]*types.Group,
-	accountID string, initiatorUser, update *types.User, addIfNotExists bool, settings *types.Settings) (bool, *types.User, []*nbpeer.Peer, []func(), error) {
+	accountID, initiatorUserId string, initiatorUser, update *types.User, addIfNotExists bool, settings *types.Settings) (bool, *types.User, []*nbpeer.Peer, []func(), error) {
 
 	if update == nil {
 		return false, nil, nil, nil, status.Errorf(status.InvalidArgument, "provided user update is nil")
@@ -653,10 +657,12 @@ func (am *DefaultAccountManager) processUserUpdate(ctx context.Context, transact
 	updatedUser.Issued = update.Issued
 	updatedUser.IntegrationReference = update.IntegrationReference
 
-	transferredOwnerRole, err := handleOwnerRoleTransfer(ctx, transaction, initiatorUser, update)
+	var transferredOwnerRole bool
+	result, err := handleOwnerRoleTransfer(ctx, transaction, initiatorUser, update)
 	if err != nil {
 		return false, nil, nil, nil, err
 	}
+	transferredOwnerRole = result
 
 	userPeers, err := transaction.GetUserPeers(ctx, store.LockingStrengthUpdate, updatedUser.AccountID, update.Id)
 	if err != nil {
@@ -676,13 +682,13 @@ func (am *DefaultAccountManager) processUserUpdate(ctx context.Context, transact
 			return false, nil, nil, nil, fmt.Errorf("error modifying user peers in groups: %w", err)
 		}
 
-		if err = transaction.SaveGroups(ctx, store.LockingStrengthUpdate, updatedGroups); err != nil {
+		if err = transaction.SaveGroups(ctx, store.LockingStrengthUpdate, accountID, updatedGroups); err != nil {
 			return false, nil, nil, nil, fmt.Errorf("error saving groups: %w", err)
 		}
 	}
 
 	updateAccountPeers := len(userPeers) > 0
-	userEventsToAdd := am.prepareUserUpdateEvents(ctx, updatedUser.AccountID, initiatorUser.Id, oldUser, updatedUser, transferredOwnerRole)
+	userEventsToAdd := am.prepareUserUpdateEvents(ctx, updatedUser.AccountID, initiatorUserId, oldUser, updatedUser, transferredOwnerRole)
 
 	return updateAccountPeers, updatedUser, peersToExpire, userEventsToAdd, nil
 }
@@ -709,7 +715,7 @@ func getUserOrCreateIfNotExists(ctx context.Context, transaction store.Store, ac
 }
 
 func handleOwnerRoleTransfer(ctx context.Context, transaction store.Store, initiatorUser, update *types.User) (bool, error) {
-	if initiatorUser.Role == types.UserRoleOwner && initiatorUser.Id != update.Id && update.Role == types.UserRoleOwner {
+	if initiatorUser != nil && initiatorUser.Role == types.UserRoleOwner && initiatorUser.Id != update.Id && update.Role == types.UserRoleOwner {
 		newInitiatorUser := initiatorUser.Copy()
 		newInitiatorUser.Role = types.UserRoleAdmin
 
@@ -737,6 +743,10 @@ func (am *DefaultAccountManager) getUserInfo(ctx context.Context, user *types.Us
 
 // validateUserUpdate validates the update operation for a user.
 func validateUserUpdate(groupsMap map[string]*types.Group, initiatorUser, oldUser, update *types.User) error {
+	if initiatorUser == nil {
+		return nil
+	}
+
 	// @todo double check these
 	if initiatorUser.HasAdminPower() && initiatorUser.Id == update.Id && oldUser.Blocked != update.Blocked {
 		return status.Errorf(status.PermissionDenied, "admins can't block or unblock themselves")
@@ -818,9 +828,13 @@ func (am *DefaultAccountManager) GetUsersFromAccount(ctx context.Context, accoun
 		return nil, status.NewPermissionValidationError(err)
 	}
 
-	user, err := am.Store.GetUserByUserID(ctx, store.LockingStrengthShare, initiatorUserID)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get user: %w", err)
+	var user *types.User
+	if initiatorUserID != activity.SystemInitiator {
+		result, err := am.Store.GetUserByUserID(ctx, store.LockingStrengthShare, initiatorUserID)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get user: %w", err)
+		}
+		user = result
 	}
 
 	accountUsers := []*types.User{}
@@ -830,7 +844,7 @@ func (am *DefaultAccountManager) GetUsersFromAccount(ctx context.Context, accoun
 		if err != nil {
 			return nil, err
 		}
-	case user.AccountID == accountID:
+	case user != nil && user.AccountID == accountID:
 		accountUsers = append(accountUsers, user)
 	default:
 		return map[string]*types.UserInfo{}, nil