Merge main into feature/affected-peers

Resolve conflicts keeping affected-peers logic while adopting
UpdateReason parameter from main for UpdateAccountPeers and
BufferUpdateAccountPeers signatures.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

management/internals/controllers/network_map/controller/controller.go

@@ -265,7 +265,10 @@ func (c *Controller) bufferSendUpdateAccountPeers(ctx context.Context, accountID
 
 // UpdatePeers updates all peers that belong to an account.
 // Should be called when changes have to be synced to peers.
-func (c *Controller) UpdateAccountPeers(ctx context.Context, accountID string) error {
+func (c *Controller) UpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error {
+	if c.accountManagerMetrics != nil {
+		c.accountManagerMetrics.CountUpdateAccountPeersTriggered(string(reason.Resource), string(reason.Operation))
+	}
 	return c.sendUpdateAccountPeers(ctx, accountID)
 }
 
@@ -466,9 +469,13 @@ func (c *Controller) UpdateAccountPeer(ctx context.Context, accountId string, pe
 	return nil
 }
 
-func (c *Controller) BufferUpdateAccountPeers(ctx context.Context, accountID string) error {
+func (c *Controller) BufferUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error {
 	log.WithContext(ctx).Tracef("buffer updating peers for account %s from %s", accountID, util.GetCallerName())
 
+	if c.accountManagerMetrics != nil {
+		c.accountManagerMetrics.CountUpdateAccountPeersTriggered(string(reason.Resource), string(reason.Operation))
+	}
+
 	bufUpd, _ := c.accountUpdateLocks.LoadOrStore(accountID, &bufferUpdate{})
 	b := bufUpd.(*bufferUpdate)
 
@@ -483,14 +490,14 @@ func (c *Controller) BufferUpdateAccountPeers(ctx context.Context, accountID str
 
 	go func() {
 		defer b.mu.Unlock()
-		_ = c.UpdateAccountPeers(ctx, accountID)
+		_ = c.sendUpdateAccountPeers(ctx, accountID)
 		if !b.update.Load() {
 			return
 		}
 		b.update.Store(false)
 		if b.next == nil {
 			b.next = time.AfterFunc(time.Duration(c.updateAccountPeersBufferInterval.Load()), func() {
-				_ = c.UpdateAccountPeers(ctx, accountID)
+				_ = c.sendUpdateAccountPeers(ctx, accountID)
 			})
 			return
 		}

management/internals/controllers/network_map/interface.go

@@ -18,11 +18,11 @@ const (
 )
 
 type Controller interface {
-	UpdateAccountPeers(ctx context.Context, accountID string) error
+	UpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error
 	UpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string) error
 	BufferUpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string) error
 	UpdateAccountPeer(ctx context.Context, accountId string, peerId string) error
-	BufferUpdateAccountPeers(ctx context.Context, accountID string) error
+	BufferUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error
 	GetValidatedPeerWithMap(ctx context.Context, isRequiresApproval bool, accountID string, p *nbpeer.Peer) (*nbpeer.Peer, *types.NetworkMap, []*posture.Checks, int64, error)
 	GetDNSDomain(settings *types.Settings) string
 	StartWarmup(context.Context)

management/internals/controllers/network_map/interface_mock.go

@@ -44,17 +44,17 @@ func (m *MockController) EXPECT() *MockControllerMockRecorder {
 }
 
 // BufferUpdateAccountPeers mocks base method.
-func (m *MockController) BufferUpdateAccountPeers(ctx context.Context, accountID string) error {
+func (m *MockController) BufferUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "BufferUpdateAccountPeers", ctx, accountID)
+	ret := m.ctrl.Call(m, "BufferUpdateAccountPeers", ctx, accountID, reason)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // BufferUpdateAccountPeers indicates an expected call of BufferUpdateAccountPeers.
-func (mr *MockControllerMockRecorder) BufferUpdateAccountPeers(ctx, accountID any) *gomock.Call {
+func (mr *MockControllerMockRecorder) BufferUpdateAccountPeers(ctx, accountID, reason any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BufferUpdateAccountPeers", reflect.TypeOf((*MockController)(nil).BufferUpdateAccountPeers), ctx, accountID)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BufferUpdateAccountPeers", reflect.TypeOf((*MockController)(nil).BufferUpdateAccountPeers), ctx, accountID, reason)
 }
 
 // BufferUpdateAffectedPeers mocks base method.
@@ -252,17 +252,17 @@ func (mr *MockControllerMockRecorder) UpdateAccountPeer(ctx, accountId, peerId a
 }
 
 // UpdateAccountPeers mocks base method.
-func (m *MockController) UpdateAccountPeers(ctx context.Context, accountID string) error {
+func (m *MockController) UpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UpdateAccountPeers", ctx, accountID)
+	ret := m.ctrl.Call(m, "UpdateAccountPeers", ctx, accountID, reason)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // UpdateAccountPeers indicates an expected call of UpdateAccountPeers.
-func (mr *MockControllerMockRecorder) UpdateAccountPeers(ctx, accountID any) *gomock.Call {
+func (mr *MockControllerMockRecorder) UpdateAccountPeers(ctx, accountID, reason any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateAccountPeers", reflect.TypeOf((*MockController)(nil).UpdateAccountPeers), ctx, accountID)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateAccountPeers", reflect.TypeOf((*MockController)(nil).UpdateAccountPeers), ctx, accountID, reason)
 }
 
 // UpdateAffectedPeers mocks base method.

management/internals/modules/peers/ephemeral/manager/ephemeral_test.go

@@ -62,7 +62,7 @@ func (a *MockAccountManager) GetDeletePeerCalls() int {
 	return a.deletePeerCalls
 }
 
-func (a *MockAccountManager) BufferUpdateAccountPeers(ctx context.Context, accountID string) {
+func (a *MockAccountManager) BufferUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	if a.bufferUpdateCalls == nil {
@@ -248,7 +248,7 @@ func TestCleanupSchedulingBehaviorIsBatched(t *testing.T) {
 					return err
 				}
 			}
-			mockAM.BufferUpdateAccountPeers(ctx, accountID)
+			mockAM.BufferUpdateAccountPeers(ctx, accountID, types.UpdateReason{})
 			return nil
 		}).
 		Times(1)

management/internals/modules/peers/manager.go

@@ -20,6 +20,7 @@ import (
 	"github.com/netbirdio/netbird/management/server/permissions/modules"
 	"github.com/netbirdio/netbird/management/server/permissions/operations"
 	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/types"
 	"github.com/netbirdio/netbird/shared/management/status"
 )
 
@@ -178,7 +179,7 @@ func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs
 		}
 	}
 
-	m.accountManager.UpdateAccountPeers(ctx, accountID)
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete})
 
 	return nil
 }

management/internals/modules/reverseproxy/service/manager/l4_port_test.go

@@ -85,7 +85,7 @@ func setupL4Test(t *testing.T, customPortsSupported *bool) (*Manager, store.Stor
 
 	accountMgr := &mock_server.MockAccountManager{
 		StoreEventFunc:         func(_ context.Context, _, _, _ string, _ activity.ActivityDescriber, _ map[string]any) {},
-		UpdateAccountPeersFunc: func(_ context.Context, _ string) {},
+		UpdateAccountPeersFunc: func(_ context.Context, _ string, _ types.UpdateReason) {},
 		GetGroupByNameFunc: func(ctx context.Context, groupName, accountID, userID string) (*types.Group, error) {
 			return testStore.GetGroupByName(ctx, store.LockingStrengthNone, accountID, groupName)
 		},

management/internals/modules/reverseproxy/service/manager/manager.go

@@ -25,6 +25,7 @@ import (
 	"github.com/netbirdio/netbird/management/server/permissions/modules"
 	"github.com/netbirdio/netbird/management/server/permissions/operations"
 	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/types"
 	"github.com/netbirdio/netbird/shared/management/status"
 )
 
@@ -231,7 +232,7 @@ func (m *Manager) CreateService(ctx context.Context, accountID, userID string, s
 
 	m.proxyController.SendServiceUpdateToCluster(ctx, accountID, s.ToProtoMapping(service.Create, "", m.proxyController.GetOIDCValidationConfig()), s.ProxyCluster)
 
-	m.accountManager.UpdateAccountPeers(ctx, accountID)
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceService, Operation: types.UpdateOperationCreate})
 
 	return s, nil
 }
@@ -515,7 +516,7 @@ func (m *Manager) UpdateService(ctx context.Context, accountID, userID string, s
 	}
 
 	m.sendServiceUpdateNotifications(ctx, accountID, service, updateInfo)
-	m.accountManager.UpdateAccountPeers(ctx, accountID)
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceService, Operation: types.UpdateOperationUpdate})
 
 	return service, nil
 }
@@ -819,7 +820,7 @@ func (m *Manager) DeleteService(ctx context.Context, accountID, userID, serviceI
 
 	m.proxyController.SendServiceUpdateToCluster(ctx, accountID, s.ToProtoMapping(service.Delete, "", m.proxyController.GetOIDCValidationConfig()), s.ProxyCluster)
 
-	m.accountManager.UpdateAccountPeers(ctx, accountID)
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceService, Operation: types.UpdateOperationDelete})
 
 	return nil
 }
@@ -860,7 +861,7 @@ func (m *Manager) DeleteAllServices(ctx context.Context, accountID, userID strin
 		m.proxyController.SendServiceUpdateToCluster(ctx, accountID, svc.ToProtoMapping(service.Delete, "", oidcCfg), svc.ProxyCluster)
 	}
 
-	m.accountManager.UpdateAccountPeers(ctx, accountID)
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceService, Operation: types.UpdateOperationDelete})
 
 	return nil
 }
@@ -916,7 +917,7 @@ func (m *Manager) ReloadService(ctx context.Context, accountID, serviceID string
 
 	m.proxyController.SendServiceUpdateToCluster(ctx, accountID, s.ToProtoMapping(service.Update, "", m.proxyController.GetOIDCValidationConfig()), s.ProxyCluster)
 
-	m.accountManager.UpdateAccountPeers(ctx, accountID)
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceService, Operation: types.UpdateOperationUpdate})
 
 	return nil
 }
@@ -1098,7 +1099,7 @@ func (m *Manager) CreateServiceFromPeer(ctx context.Context, accountID, peerID s
 	}
 
 	m.proxyController.SendServiceUpdateToCluster(ctx, accountID, svc.ToProtoMapping(service.Create, "", m.proxyController.GetOIDCValidationConfig()), svc.ProxyCluster)
-	m.accountManager.UpdateAccountPeers(ctx, accountID)
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceService, Operation: types.UpdateOperationCreate})
 
 	serviceURL := "https://" + svc.Domain
 	if service.IsL4Protocol(svc.Mode) {
@@ -1210,7 +1211,7 @@ func (m *Manager) deletePeerService(ctx context.Context, accountID, peerID, serv
 
 	m.proxyController.SendServiceUpdateToCluster(ctx, accountID, svc.ToProtoMapping(service.Delete, "", m.proxyController.GetOIDCValidationConfig()), svc.ProxyCluster)
 
-	m.accountManager.UpdateAccountPeers(ctx, accountID)
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceService, Operation: types.UpdateOperationDelete})
 
 	return nil
 }
@@ -1261,7 +1262,7 @@ func (m *Manager) deleteExpiredPeerService(ctx context.Context, accountID, peerI
 	meta := addPeerInfoToEventMeta(svc.EventMeta(), peer)
 	m.accountManager.StoreEvent(ctx, peerID, serviceID, accountID, activity.PeerServiceExposeExpired, meta)
 	m.proxyController.SendServiceUpdateToCluster(ctx, accountID, svc.ToProtoMapping(service.Delete, "", m.proxyController.GetOIDCValidationConfig()), svc.ProxyCluster)
-	m.accountManager.UpdateAccountPeers(ctx, accountID)
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceService, Operation: types.UpdateOperationDelete})
 
 	return nil
 }

management/internals/modules/reverseproxy/service/manager/manager_test.go

@@ -447,7 +447,7 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) {
 			StoreEventFunc: func(_ context.Context, _, _, _ string, activityID activity.ActivityDescriber, _ map[string]any) {
 				storedActivity = activityID.(activity.Activity)
 			},
-			UpdateAccountPeersFunc: func(_ context.Context, _ string) {},
+			UpdateAccountPeersFunc: func(_ context.Context, _ string, _ types.UpdateReason) {},
 		}
 
 		mockStore.EXPECT().
@@ -549,7 +549,7 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) {
 			StoreEventFunc: func(_ context.Context, _, _, _ string, activityID activity.ActivityDescriber, _ map[string]any) {
 				storedActivity = activityID.(activity.Activity)
 			},
-			UpdateAccountPeersFunc: func(_ context.Context, _ string) {},
+			UpdateAccountPeersFunc: func(_ context.Context, _ string, _ types.UpdateReason) {},
 		}
 
 		mockStore.EXPECT().
@@ -593,7 +593,7 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) {
 			StoreEventFunc: func(_ context.Context, _, _, _ string, _ activity.ActivityDescriber, meta map[string]any) {
 				storedMeta = meta
 			},
-			UpdateAccountPeersFunc: func(_ context.Context, _ string) {},
+			UpdateAccountPeersFunc: func(_ context.Context, _ string, _ types.UpdateReason) {},
 		}
 
 		mockStore.EXPECT().
@@ -704,7 +704,7 @@ func setupIntegrationTest(t *testing.T) (*Manager, store.Store) {
 
 	accountMgr := &mock_server.MockAccountManager{
 		StoreEventFunc:         func(_ context.Context, _, _, _ string, _ activity.ActivityDescriber, _ map[string]any) {},
-		UpdateAccountPeersFunc: func(_ context.Context, _ string) {},
+		UpdateAccountPeersFunc: func(_ context.Context, _ string, _ types.UpdateReason) {},
 		GetGroupByNameFunc: func(ctx context.Context, groupName, accountID, userID string) (*types.Group, error) {
 			return testStore.GetGroupByName(ctx, store.LockingStrengthNone, accountID, groupName)
 		},
@@ -1173,7 +1173,7 @@ func TestDeleteService_DeletesTargets(t *testing.T) {
 	mockAcct.EXPECT().
 		StoreEvent(ctx, userID, service.ID, accountID, activity.ServiceDeleted, gomock.Any())
 	mockAcct.EXPECT().
-		UpdateAccountPeers(ctx, accountID)
+		UpdateAccountPeers(ctx, accountID, gomock.Any())
 
 	err = mgr.DeleteService(ctx, accountID, userID, service.ID)
 	require.NoError(t, err)

management/internals/modules/zones/manager/manager.go

@@ -11,6 +11,7 @@ import (
 	"github.com/netbirdio/netbird/management/server/permissions/modules"
 	"github.com/netbirdio/netbird/management/server/permissions/operations"
 	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/types"
 	"github.com/netbirdio/netbird/shared/management/status"
 )
 
@@ -144,7 +145,7 @@ func (m *managerImpl) UpdateZone(ctx context.Context, accountID, userID string,
 
 	m.accountManager.StoreEvent(ctx, userID, zone.ID, accountID, activity.DNSZoneUpdated, zone.EventMeta())
 
-	go m.accountManager.UpdateAccountPeers(ctx, accountID)
+	go m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceZone, Operation: types.UpdateOperationUpdate})
 
 	return zone, nil
 }
@@ -206,7 +207,7 @@ func (m *managerImpl) DeleteZone(ctx context.Context, accountID, userID, zoneID
 		event()
 	}
 
-	go m.accountManager.UpdateAccountPeers(ctx, accountID)
+	go m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceZone, Operation: types.UpdateOperationDelete})
 
 	return nil
 }

management/internals/modules/zones/records/manager/manager.go

@@ -13,6 +13,7 @@ import (
 	"github.com/netbirdio/netbird/management/server/permissions/modules"
 	"github.com/netbirdio/netbird/management/server/permissions/operations"
 	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/types"
 	"github.com/netbirdio/netbird/shared/management/status"
 )
 
@@ -95,7 +96,7 @@ func (m *managerImpl) CreateRecord(ctx context.Context, accountID, userID, zoneI
 	meta := record.EventMeta(zone.ID, zone.Name)
 	m.accountManager.StoreEvent(ctx, userID, record.ID, accountID, activity.DNSRecordCreated, meta)
 
-	go m.accountManager.UpdateAccountPeers(ctx, accountID)
+	go m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceZoneRecord, Operation: types.UpdateOperationCreate})
 
 	return record, nil
 }
@@ -154,7 +155,7 @@ func (m *managerImpl) UpdateRecord(ctx context.Context, accountID, userID, zoneI
 	meta := record.EventMeta(zone.ID, zone.Name)
 	m.accountManager.StoreEvent(ctx, userID, record.ID, accountID, activity.DNSRecordUpdated, meta)
 
-	go m.accountManager.UpdateAccountPeers(ctx, accountID)
+	go m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceZoneRecord, Operation: types.UpdateOperationUpdate})
 
 	return record, nil
 }
@@ -201,7 +202,7 @@ func (m *managerImpl) DeleteRecord(ctx context.Context, accountID, userID, zoneI
 	meta := record.EventMeta(zone.ID, zone.Name)
 	m.accountManager.StoreEvent(ctx, userID, recordID, accountID, activity.DNSRecordDeleted, meta)
 
-	go m.accountManager.UpdateAccountPeers(ctx, accountID)
+	go m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourceZoneRecord, Operation: types.UpdateOperationDelete})
 
 	return nil
 }

management/internals/server/boot.go

@@ -173,7 +173,7 @@ func (s *BaseServer) GRPCServer() *grpc.Server {
 		}
 
 		gRPCAPIHandler := grpc.NewServer(gRPCOpts...)
-		srv, err := nbgrpc.NewServer(s.Config, s.AccountManager(), s.SettingsManager(), s.JobManager(), s.SecretsManager(), s.Metrics(), s.AuthManager(), s.IntegratedValidator(), s.NetworkMapController(), s.OAuthConfigProvider())
+		srv, err := nbgrpc.NewServer(s.Config, s.AccountManager(), s.SettingsManager(), s.JobManager(), s.SecretsManager(), s.Metrics(), s.AuthManager(), s.IntegratedValidator(), s.NetworkMapController(), s.OAuthConfigProvider(), s.SessionStore())
 		if err != nil {
 			log.Fatalf("failed to create management server: %v", err)
 		}

management/internals/server/controllers.go

@@ -6,6 +6,7 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	"github.com/netbirdio/management-integrations/integrations"
+
 	"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
 	proxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy/manager"
 
@@ -66,6 +67,12 @@ func (s *BaseServer) SecretsManager() grpc.SecretsManager {
 	})
 }
 
+func (s *BaseServer) SessionStore() *auth.SessionStore {
+	return Create(s, func() *auth.SessionStore {
+		return auth.NewSessionStore(s.CacheStore())
+	})
+}
+
 func (s *BaseServer) AuthManager() auth.Manager {
 	audiences := s.Config.GetAuthAudiences()
 	audience := s.Config.HttpConfig.AuthAudience

management/internals/shared/grpc/server.go

@@ -14,6 +14,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	jwtv5 "github.com/golang-jwt/jwt/v5"
 	pb "github.com/golang/protobuf/proto" // nolint
 	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
@@ -67,6 +68,7 @@ type Server struct {
 	appMetrics     telemetry.AppMetrics
 	peerLocks      sync.Map
 	authManager    auth.Manager
+	sessionStore   *auth.SessionStore
 
 	logBlockedPeers          bool
 	blockPeersWithSameConfig bool
@@ -98,6 +100,7 @@ func NewServer(
 	integratedPeerValidator integrated_validator.IntegratedValidator,
 	networkMapController network_map.Controller,
 	oAuthConfigProvider idp.OAuthConfigProvider,
+	sessionStore *auth.SessionStore,
 ) (*Server, error) {
 	if appMetrics != nil {
 		// update gauge based on number of connected peers which is equal to open gRPC streams
@@ -140,6 +143,7 @@ func NewServer(
 		integratedPeerValidator:  integratedPeerValidator,
 		networkMapController:     networkMapController,
 		oAuthConfigProvider:      oAuthConfigProvider,
+		sessionStore:             sessionStore,
 
 		loginFilter: newLoginFilter(),
 
@@ -535,7 +539,7 @@ func (s *Server) cancelPeerRoutinesWithoutLock(ctx context.Context, accountID st
 	log.WithContext(ctx).Debugf("peer %s has been disconnected", peer.Key)
 }
 
-func (s *Server) validateToken(ctx context.Context, jwtToken string) (string, error) {
+func (s *Server) validateToken(ctx context.Context, peerKey, jwtToken string) (string, error) {
 	if s.authManager == nil {
 		return "", status.Errorf(codes.Internal, "missing auth manager")
 	}
@@ -545,6 +549,10 @@ func (s *Server) validateToken(ctx context.Context, jwtToken string) (string, er
 		return "", status.Errorf(codes.InvalidArgument, "invalid jwt token, err: %v", err)
 	}
 
+	if err := s.claimLoginToken(ctx, peerKey, jwtToken, token); err != nil {
+		return "", err
+	}
+
 	// we need to call this method because if user is new, we will automatically add it to existing or create a new account
 	accountId, _, err := s.accountManager.GetAccountIDFromUserAuth(ctx, userAuth)
 	if err != nil {
@@ -828,6 +836,31 @@ func (s *Server) prepareLoginResponse(ctx context.Context, peer *nbpeer.Peer, ne
 	return loginResp, nil
 }
 
+func (s *Server) claimLoginToken(ctx context.Context, peerKey, jwtToken string, token *jwtv5.Token) error {
+	if s.sessionStore == nil || token == nil {
+		return nil
+	}
+
+	exp, err := token.Claims.GetExpirationTime()
+	if err != nil || exp == nil {
+		log.WithContext(ctx).Warnf("JWT has no usable exp claim for peer %s", peerKey)
+		return status.Error(codes.Unauthenticated, "jwt token has no expiration")
+	}
+
+	err = s.sessionStore.RegisterToken(ctx, jwtToken, exp.Time)
+	if err == nil {
+		return nil
+	}
+
+	if errors.Is(err, auth.ErrTokenAlreadyUsed) || errors.Is(err, auth.ErrTokenExpired) {
+		log.WithContext(ctx).Warnf("%v for peer %s", err, peerKey)
+		return status.Error(codes.Unauthenticated, err.Error())
+	}
+
+	log.WithContext(ctx).Warnf("failed to claim JWT for peer %s: %v", peerKey, err)
+	return status.Error(codes.Unavailable, "failed to claim jwt token")
+}
+
 // processJwtToken validates the existence of a JWT token in the login request, and returns the corresponding user ID if
 // the token is valid.
 //
@@ -838,7 +871,7 @@ func (s *Server) processJwtToken(ctx context.Context, loginReq *proto.LoginReque
 	if loginReq.GetJwtToken() != "" {
 		var err error
 		for i := 0; i < 3; i++ {
-			userID, err = s.validateToken(ctx, loginReq.GetJwtToken())
+			userID, err = s.validateToken(ctx, peerKey.String(), loginReq.GetJwtToken())
 			if err == nil {
 				break
 			}